{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 88 启动协程（Goroutine）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题               | 核心能力                                        | 关键 API / 技巧                              | 使用提示                                          |\n",
    "| ------------------ | ----------------------------------------------- | -------------------------------------------- | ------------------------------------------------- |\n",
    "| Goroutine 基本概念 | Go 中的并发执行单元，轻量级、由 Go 运行时调度   | 使用 `go` 关键字启动                         | 启动后与主协程并发执行；执行顺序非确定            |\n",
    "| CPU 信息获取       | 查看逻辑处理器数量                              | `runtime.NumCPU()`                           | 超线程机器返回逻辑核数；可用于资源评估            |\n",
    "| 核心数限制         | 控制可并行执行的最大 P 数量                     | `runtime.GOMAXPROCS(n)`                      | 默认值为全部逻辑核；可动态调整限制                |\n",
    "| 协程数量观测       | 获取当前存活协程数量                            | `runtime.NumGoroutine()`                     | 已结束协程不会统计；主协程退出子协程随之退出      |\n",
    "| 函数返回值         | 协程函数返回值无法直接获取                      | 建议通过通道、`WaitGroup`、`sync` 包传递结果 | 忽略返回值编译器不会报错，但逻辑可能出错          |\n",
    "| 启动方式           | 支持命名函数、匿名函数、函数变量                | `go someFunc()`、`go func(){}`、`go fnVar()` | 匿名函数可直接捕获上下文变量                      |\n",
    "| 调度模型           | GMP 模型：G=goroutine，M=OS thread，P=processor | ——                                           | 大量 goroutine 会排队等待调度，非同一时刻全部运行 |\n",
    "| 生命周期控制       | 防止主协程提前退出                              | `time.Sleep`、`sync.WaitGroup`、`context`    | 正式代码推荐同步原语，避免使用 `Sleep`            |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 基础协程与 WaitGroup 协调"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "逻辑处理器数量: 16\n",
      "启动前协程数: 1\n",
      "协程已启动，当前存活数: 4\n",
      "Add(2, 3) = 5\n",
      "Add(1, 2) = 3\n",
      "Add(0, 1) = 1\n",
      "协程执行完毕，当前存活数: 1\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func Add(a, b int) {\n",
    "\tfmt.Printf(\"Add(%d, %d) = %d\\n\", a, b, a+b)\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tfmt.Println(\"逻辑处理器数量:\", runtime.NumCPU())\n",
    "\tfmt.Println(\"启动前协程数:\", runtime.NumGoroutine())\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\tfor i := 0; i < 3; i++ {\n",
    "\t\twg.Add(1)\n",
    "\t\tgo func(x int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tAdd(x, x+1)\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\tfmt.Println(\"协程已启动，当前存活数:\", runtime.NumGoroutine())\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"协程执行完毕，当前存活数:\", runtime.NumGoroutine())\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 控制最大并行核数（runtime.GOMAXPROCS）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "逻辑核总数: 16，原GOMAXPROCS: 16，现设置为: 8\n",
      "worker(6) 完成，sum=12499997500000\n",
      "worker(5) 完成，sum=12499997500000\n",
      "worker(0) 完成，sum=12499997500000\n",
      "worker(3) 完成，sum=12499997500000\n",
      "worker(2) 完成，sum=12499997500000\n",
      "worker(1) 完成，sum=12499997500000\n",
      "worker(11) 完成，sum=12499997500000\n",
      "worker(8) 完成，sum=12499997500000\n",
      "worker(12) 完成，sum=12499997500000\n",
      "worker(4) 完成，sum=12499997500000\n",
      "worker(7) 完成，sum=12499997500000\n",
      "worker(9) 完成，sum=12499997500000\n",
      "worker(10) 完成，sum=12499997500000\n",
      "worker(14) 完成，sum=12499997500000\n",
      "worker(15) 完成，sum=12499997500000\n",
      "worker(13) 完成，sum=12499997500000\n",
      "全部任务完成\n"
     ]
    }
   ],
   "source": [
    "// 示范限制进程可用核数。运行时可观察协程调度与并行度变化。若不希望循环打印，可移除 Tick 协程。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\ttotalCPU := runtime.NumCPU()\n",
    "\tprev := runtime.GOMAXPROCS(totalCPU / 2)\n",
    "\tfmt.Printf(\"逻辑核总数: %d，原GOMAXPROCS: %d，现设置为: %d\\n\", totalCPU, prev, totalCPU/2)\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\tfor i := 0; i < totalCPU; i++ {\n",
    "\t\twg.Add(1)\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tsum := 0\n",
    "\t\t\tfor j := 0; j < 5_000_000; j++ {\n",
    "\t\t\t\tsum += j\n",
    "\t\t\t}\n",
    "\t\t\tfmt.Printf(\"worker(%d) 完成，sum=%d\\n\", id, sum)\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\tgo func() {\n",
    "\t\tfor range time.Tick(200 * time.Millisecond) {\n",
    "\t\t\tfmt.Println(\"当前协程数:\", runtime.NumGoroutine())\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"全部任务完成\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 匿名函数启动协程并安全捕获变量"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hello, Carol\n",
      "Hello, Bob\n",
      "Hello, Alice\n"
     ]
    }
   ],
   "source": [
    "// 注意循环变量要通过形参捕获，避免协程读取到同一地址导致输出重复\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar wg sync.WaitGroup\n",
    "\tnames := []string{\"Alice\", \"Bob\", \"Carol\"}\n",
    "\n",
    "\tfor _, name := range names {\n",
    "\t\twg.Add(1)\n",
    "\t\tgo func(n string) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tfmt.Println(\"Hello,\", n)\n",
    "\t\t}(name)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 函数变量与协程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "add(3, 7) => 10\n",
      "add(2, 5) => 7\n",
      "add(1, 9) => 10\n"
     ]
    }
   ],
   "source": [
    "// 通过函数变量封装逻辑后，直接使用 go add(...) 启动协程，便于按需替换实现。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "var add = func(a, b int) {\n",
    "\tfmt.Printf(\"add(%d, %d) => %d\\n\", a, b, a+b)\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tvar wg sync.WaitGroup\n",
    "\ttasks := [][2]int{{1, 9}, {2, 5}, {3, 7}}\n",
    "\n",
    "\tfor _, t := range tasks {\n",
    "\t\twg.Add(1)\n",
    "\t\tgo func(x, y int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tadd(x, y)\n",
    "\t\t}(t[0], t[1])\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 主协程提前退出示例（非建议做法）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "主协程立即退出\n"
     ]
    }
   ],
   "source": [
    "// 运行后通常只打印“主协程立即退出”。主协程结束会终止整个进程，子协程没有机会执行。真实业务必须使用 WaitGroup、通道或其他同步手段确保协程安全收尾。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(500 * time.Millisecond)\n",
    "\t\tfmt.Println(\"子协程完成\")\n",
    "\t}()\n",
    "\tfmt.Println(\"主协程立即退出\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 89 协程的生命周期与`WaitGroup`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题               | 核心概念                                | 关键 API                 | 注意事项                                    |\n",
    "| ------------------ | --------------------------------------- | ------------------------ | ------------------------------------------- |\n",
    "| 协程独立性         | 除 `main` 外，所有 goroutine 均独立存在 | `go` 关键字              | 任何子/孙协程完成次序不可预测；互不依赖     |\n",
    "| 主协程特殊性       | `main` 结束会强制终止所有存活协程       | `main` 函数返回          | 若主协程提前退出，未完成的协程立即被杀死    |\n",
    "| 协程数量监控       | 查看当前活跃 goroutine                  | `runtime.NumGoroutine()` | 便于调试、观察生命周期变化                  |\n",
    "| `WaitGroup` 计数器 | 等待一组协程完成                        | `Add` / `Done` / `Wait`  | `Done` 必须与 `Add` 配对，推荐 `defer`      |\n",
    "| 协程启动模式       | 命名函数、匿名函数、函数变量            | `go fn()`、`go func(){}` | 返回值无法直接获取，需配合 channel/共享状态 |\n",
    "| 调度随机性         | 调度顺序由运行时决定                    | ——                       | 打印顺序常乱序；勿依赖执行顺序              |\n",
    "| 资源约束           | 合理限制并发度                          | `runtime.GOMAXPROCS`     | 避免过多 goroutine 造成资源竞争             |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 协程生命周期（父、子、孙协程独立执行）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "main start\n",
      "child begin\n",
      "grandson begin\n",
      "goroutines: 3\n",
      "child finish\n",
      "grandson finish\n",
      "main exit\n"
     ]
    }
   ],
   "source": [
    "// child 结束后 grandson 仍继续运行。\n",
    "// 将 main 中 Sleep 改为 2 * time.Second 会导致 grandson 尚未打印 finish 就被中断，体现主协程的特殊性。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func grandson() {\n",
    "\tfmt.Println(\"grandson begin\")\n",
    "\tfmt.Println(\"goroutines:\", runtime.NumGoroutine())\n",
    "\ttime.Sleep(3 * time.Second)\n",
    "\tfmt.Println(\"grandson finish\")\n",
    "}\n",
    "\n",
    "func child() {\n",
    "\tfmt.Println(\"child begin\")\n",
    "\tgo grandson()\n",
    "\ttime.Sleep(100 * time.Millisecond)\n",
    "\tfmt.Println(\"child finish\")\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tfmt.Println(\"main start\")\n",
    "\tgo child()\n",
    "\ttime.Sleep(5 * time.Second) // 留足时间观察孙协程完成\n",
    "\tfmt.Println(\"main exit\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `WaitGroup` 等待多协程收尾"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `worker` 完成顺序随机。\n",
    "- `wg.Wait()` 保障所有协程结束后主协程才退出。\n",
    "- 使用 `defer wg.Done()` 确保即便协程内部出现 `panic` 前已 `Done` 也能计数递减。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "活跃协程数: 11\n",
      "worker 8 done\n",
      "worker 2 done\n",
      "worker 4 done\n",
      "worker 9 done\n",
      "worker 6 done\n",
      "worker 7 done\n",
      "worker 5 done\n",
      "worker 1 done\n",
      "worker 3 done\n",
      "worker 0 done\n",
      "全部完成，活跃协程数: 1\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tconst workers = 10\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()        // 对应 Add(+1)\n",
    "\t\t\ttime.Sleep(10 * time.Millisecond)\n",
    "\t\t\tfmt.Printf(\"worker %d done\\n\", id)\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\tfmt.Println(\"活跃协程数:\", runtime.NumGoroutine())\n",
    "\twg.Wait() // 等待计数归零\n",
    "\tfmt.Println(\"全部完成，活跃协程数:\", runtime.NumGoroutine())\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### WaitGroup 嵌套场景（父协程等待孙协程）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 通过内外层 `WaitGroup` 分别管理不同协程集合，保证退出顺序可控。\n",
    "- 若省略内层 `WaitGroup`，`child` 在 `grandson` 完成前可能结束，提示 `WaitGroup` 需覆盖所有需要等待的协程。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "child running\n",
      "grandson done\n",
      "child done\n",
      "main exit\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(1)\n",
    "\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tvar inner sync.WaitGroup\n",
    "\t\tinner.Add(1)\n",
    "\t\tgo func() {\n",
    "\t\t\tdefer inner.Done()\n",
    "\t\t\ttime.Sleep(500 * time.Millisecond)\n",
    "\t\t\tfmt.Println(\"grandson done\")\n",
    "\t\t}()\n",
    "\t\tfmt.Println(\"child running\")\n",
    "\t\tinner.Wait() // child 确保孙协程完成\n",
    "\t\tfmt.Println(\"child done\")\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"main exit\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 90 并发安全性与原子操作"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题                      | 核心概念                                                     | 关键 API / 工具                                              | 使用提示                                           |\n",
    "| ------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ | -------------------------------------------------- |\n",
    "| 资源竞争 (Race Condition) | 多个协程并发读写同一内存，导致结果不可预测                   | `go run -race`                                               | 通过竞态检测器及时发现数据竞争                     |\n",
    "| 非原子操作                | `n++` 会被拆解为「读 → 加 → 写」三步，无法保证一致性         | ——                                                           | 并发中会产生脏写，最终结果小于预期                 |\n",
    "| 原子操作基本函数          | 原子加减、交换、比较交换、加载/存储                          | `atomic.AddInt32`、`atomic.CompareAndSwapInt32`、`atomic.LoadInt32` | 操作不可分割，保证同一时刻仅有一个协程修改目标地址 |\n",
    "| `atomic.Value`            | 并发安全地存储任意类型数据（需保持类型一致）                 | `val.Store(v)`、`val.Load()`                                 | 适合热更新配置、共享缓存等场景                     |\n",
    "| 同步 vs 原子              | 锁 (`sync.Mutex`) 提供更灵活的互斥控制；原子操作更轻量但仅适用于简单场景 | `sync.Mutex`、`atomic`                                       | 复杂临界区优先使用锁，简单计数/标志使用原子        |\n",
    "| 重置与测试                | 每轮实验前需重置共享变量；通过 WaitGroup 等待协程完成        | `wg.Add`/`wg.Done`/`wg.Wait`                                 | 确保实验条件一致、程序不会提前退出                 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 非原子 n++ 引发的脏写现象"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "非原子加法期望值=1000，实际值=960\n",
      "多运行几次会看到 counter < goroutines，说明发生了脏写。\n",
      "建议使用 go run -race main.go 检测数据竞争。\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var (\n",
    "\tcounter int32          // 共享变量：多个协程会同时访问\n",
    "\twg      sync.WaitGroup // 用于等待所有协程结束\n",
    ")\n",
    "\n",
    "func incrementNonAtomic() {\n",
    "\tdefer wg.Done()\n",
    "\n",
    "\t// 模拟更明显的冲突：休眠让调度器有机会切换协程\n",
    "\ttime.Sleep(time.Microsecond)\n",
    "\n",
    "\t// 非原子操作：实际上拆解为「读 → 加 → 写」三步\n",
    "\tcounter++ // counter = counter + 1\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst goroutines = 1000\n",
    "\tcounter = 0\n",
    "\twg.Add(goroutines)\n",
    "\n",
    "\tfor i := 0; i < goroutines; i++ {\n",
    "\t\tgo incrementNonAtomic()\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Printf(\"非原子加法期望值=%d，实际值=%d\\n\", goroutines, counter)\n",
    "\tfmt.Println(\"多运行几次会看到 counter < goroutines，说明发生了脏写。\")\n",
    "\tfmt.Println(\"建议使用 go run -race main.go 检测数据竞争。\")\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "==================\n",
      "WARNING: DATA RACE\n",
      "Read at 0x0000006035e8 by goroutine 10:\n",
      "  main.incrementNonAtomic()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:21 +0x7e\n",
      "\n",
      "Previous write at 0x0000006035e8 by goroutine 7:\n",
      "  main.incrementNonAtomic()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:21 +0x94\n",
      "\n",
      "Goroutine 10 (running) created at:\n",
      "  main.main()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:30 +0x64\n",
      "\n",
      "Goroutine 7 (finished) created at:\n",
      "  main.main()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:30 +0x64\n",
      "==================\n",
      "==================\n",
      "WARNING: DATA RACE\n",
      "Write at 0x0000006035e8 by goroutine 10:\n",
      "  main.incrementNonAtomic()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:21 +0x94\n",
      "\n",
      "Previous write at 0x0000006035e8 by goroutine 8:\n",
      "  main.incrementNonAtomic()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:21 +0x94\n",
      "\n",
      "Goroutine 10 (running) created at:\n",
      "  main.main()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:30 +0x64\n",
      "\n",
      "Goroutine 8 (running) created at:\n",
      "  main.main()\n",
      "      /mnt/d/dev_learn/golang_from_beginner_to_mastery/04_concurrent_programming/90/main.go:30 +0x64\n",
      "==================\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "非原子加法期望值=1000，实际值=777\n",
      "多运行几次会看到 counter < goroutines，说明发生了脏写。\n",
      "建议使用 go run -race main.go 检测数据竞争。\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Found 2 data race(s)\n",
      "exit status 66\n",
      "exit status 1\n"
     ]
    }
   ],
   "source": [
    "! cd ./90 && go run -race ."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 使用 atomic.AddInt32 修复脏写问题"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "使用 atomic.AddInt32 后的计数结果=1000（始终等于 1000）\n",
      "普通读取 total=1000，atomic.LoadInt32(total)=1000\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"sync/atomic\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var (\n",
    "\ttotal int32         // 需要并发安全写入的计数器\n",
    "\twg    sync.WaitGroup\n",
    ")\n",
    "\n",
    "func incrementAtomic() {\n",
    "\tdefer wg.Done()\n",
    "\n",
    "\ttime.Sleep(time.Microsecond)\n",
    "\n",
    "\t// 原子加：整个「读 → 加 → 写」操作被封装为不可中断的指令\n",
    "\tatomic.AddInt32(&total, 1)\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst workers = 1000\n",
    "\ttotal = 0\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo incrementAtomic()\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Printf(\"使用 atomic.AddInt32 后的计数结果=%d（始终等于 %d）\\n\", total, workers)\n",
    "\n",
    "\t// 读取操作也可以直接使用普通读取；atomic.LoadInt32 更偏向需要保证顺序一致性的极端场景\n",
    "\tfmt.Printf(\"普通读取 total=%d，atomic.LoadInt32(total)=%d\\n\", total, atomic.LoadInt32(&total))\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 原子比较交换 CompareAndSwap(CAS) 控制状态流转"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "worker 4 获得执行权限\n",
      "worker 0 未抢到任务，当前状态=2\n",
      "worker 1 未抢到任务，当前状态=2\n",
      "worker 2 未抢到任务，当前状态=2\n",
      "worker 3 未抢到任务，当前状态=2\n",
      "最终状态: 2\n"
     ]
    }
   ],
   "source": [
    "// CompareAndSwap 常用于实现一次性初始化、锁、信号等场景。与互斥锁相比，CAS 避免了内核态切换，但逻辑复杂时仍建议使用 sync.Mutex。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"sync/atomic\"\n",
    ")\n",
    "\n",
    "const (\n",
    "\tstateIdle int32 = iota\n",
    "\tstateRunning\n",
    "\tstateFinished\n",
    ")\n",
    "\n",
    "var (\n",
    "\ttaskState int32 = stateIdle\n",
    "\twg        sync.WaitGroup\n",
    ")\n",
    "\n",
    "func runOnce(id int) {\n",
    "\tdefer wg.Done()\n",
    "\n",
    "\t// 只有第一个成功把状态从 Idle 改为 Running 的协程可以执行任务\n",
    "\tif atomic.CompareAndSwapInt32(&taskState, stateIdle, stateRunning) {\n",
    "\t\tfmt.Printf(\"worker %d 获得执行权限\\n\", id)\n",
    "\n",
    "\t\t// 模拟任务执行\n",
    "\t\t// ... 执行实际逻辑 ...\n",
    "\n",
    "\t\t// 标记为完成状态\n",
    "\t\tatomic.StoreInt32(&taskState, stateFinished)\n",
    "\t} else {\n",
    "\t\tfmt.Printf(\"worker %d 未抢到任务，当前状态=%d\\n\", id, atomic.LoadInt32(&taskState))\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst workers = 5\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo runOnce(i)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"最终状态:\", atomic.LoadInt32(&taskState))\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 使用 atomic.Value 并发安全地热更新配置"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "reader 1 -> version=0 msg=bootstrap\n",
      "reader 2 -> version=0 msg=bootstrap\n",
      "reader 1 -> version=0 msg=bootstrap\n",
      "reader 2 -> version=0 msg=bootstrap\n",
      "writer updated: {1 update-1}\n",
      "reader 2 -> version=1 msg=update-1\n",
      "reader 1 -> version=1 msg=update-1\n",
      "writer updated: {2 update-2}\n",
      "writer updated: {3 update-3}\n"
     ]
    }
   ],
   "source": [
    "// atomic.Value 要求存储的类型一致（首次 Store 后类型固定），适合多读少写的配置场景。若写入频繁，建议使用 sync.RWMutex。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"sync/atomic\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "type Config struct {\n",
    "\tVersion int\n",
    "\tMessage string\n",
    "}\n",
    "\n",
    "var (\n",
    "\tconfigValue atomic.Value // 保存最新配置\n",
    "\tflag        int32         // 控制更新频率\n",
    "\twg          sync.WaitGroup\n",
    ")\n",
    "\n",
    "func reader(id int) {\n",
    "\tdefer wg.Done()\n",
    "\tfor i := 0; i < 3; i++ {\n",
    "\t\tcfg := configValue.Load().(Config) // 并发安全读取\n",
    "\t\tfmt.Printf(\"reader %d -> version=%d msg=%s\\n\", id, cfg.Version, cfg.Message)\n",
    "\t\ttime.Sleep(150 * time.Millisecond)\n",
    "\t}\n",
    "}\n",
    "\n",
    "func writer() {\n",
    "\tdefer wg.Done()\n",
    "\tfor v := 1; v <= 3; v++ {\n",
    "\t\ttime.Sleep(200 * time.Millisecond)\n",
    "\n",
    "\t\t// 一次性进入临界区：确保同一时刻只有一个 writer 执行写入\n",
    "\t\tif atomic.CompareAndSwapInt32(&flag, 0, 1) {\n",
    "\t\t\tcfg := Config{\n",
    "\t\t\t\tVersion: v,\n",
    "\t\t\t\tMessage: fmt.Sprintf(\"update-%d\", v),\n",
    "\t\t\t}\n",
    "\t\t\tconfigValue.Store(cfg) // 并发安全写入\n",
    "\t\t\tfmt.Println(\"writer updated:\", cfg)\n",
    "\t\t\tatomic.StoreInt32(&flag, 0)\n",
    "\t\t}\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\t// 初始配置必须先写入，否则 Load 会 panic\n",
    "\tconfigValue.Store(Config{Version: 0, Message: \"bootstrap\"})\n",
    "\n",
    "\twg.Add(3)\n",
    "\tgo reader(1)\n",
    "\tgo reader(2)\n",
    "\tgo writer()\n",
    "\n",
    "\twg.Wait()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 91 读写锁"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题         | 核心概念                                                     | 关键 API / 类型                                             | 使用提示                                                     |\n",
    "| ------------ | ------------------------------------------------------------ | ----------------------------------------------------------- | ------------------------------------------------------------ |\n",
    "| 互斥锁基础   | `sync.Mutex` 保证同一时刻只有一个协程进入临界区              | `lock.Lock()` / `lock.Unlock()`                             | 加锁与解锁必须成对出现，推荐 `defer` 确保释放                |\n",
    "| 写锁即互斥锁 | `Mutex` 等价于“写锁”——任何时候最多一协程访问共享数据         | ——                                                          | 高并发下会串行化临界区，避免脏写                             |\n",
    "| 读写锁结构   | `sync.RWMutex` 同时提供读锁和写锁                            | `mu.RLock()` / `mu.RUnlock()` / `mu.Lock()` / `mu.Unlock()` | 读锁允许多个协程共享读，写锁只允许唯一协程                   |\n",
    "| 可重入性     | 读锁可重入（同一协程多次 `RLock`）；写锁不可重入（再次 `Lock` 会死锁） | ——                                                          | 写锁重入导致 `fatal error: all goroutines are asleep - deadlock!` |\n",
    "| 互斥规则     | 持有读锁时：其他读锁可获取，写锁阻塞；持有写锁时：所有锁请求阻塞 | ——                                                          | 读锁不阻塞读，但阻塞写；写锁阻塞一切                         |\n",
    "| 排队与公平性 | Go 的读写锁遵循先来先服务：先排队的写锁优先于后来的读锁，防止写锁饥饿 | ——                                                          | 当写锁在等待时，后续读锁也得排队，确保写锁最终能获得         |\n",
    "| 复制限制     | `RWMutex`、`Mutex` 属于非复制类型；一旦使用，禁止再复制      | ——                                                          | 在结构体中应该以指针方式传递或使用指针接收者                 |\n",
    "| 组合策略     | 读多写少：选用 `RWMutex` 提升读并发；写操作频繁：直接用 `Mutex` 简化逻辑 | ——                                                          | 读写比例决定选型，避免过度优化                               |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 用互斥锁修复计数脏写问题"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 每次递增前后都必须 `Lock()` / `Unlock()`。\n",
    "- 高并发下无论运行多少次，结果都与预期相同（1000）。\n",
    "- 若去掉锁，很容易出现小于预期的脏写结果。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "最终结果 counter=1000（预期值=1000）\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "var (\n",
    "\tlock    sync.Mutex // 互斥锁：同一时刻只有一个协程进入临界区\n",
    "\tcounter int32       // 共享计数器\n",
    ")\n",
    "\n",
    "func safeIncrement(wg *sync.WaitGroup) {\n",
    "\tdefer wg.Done()\n",
    "\n",
    "\tlock.Lock()   // 加写锁，确保下面的 counter++ 在临界区串行执行\n",
    "\tcounter++     // 临界区：对共享变量的写操作\n",
    "\tlock.Unlock() // 释放写锁，唤醒其他等待协程\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst workers = 1000\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo safeIncrement(&wg) // 并发启动 1000 个协程递增 counter\n",
    "\t}\n",
    "\n",
    "\twg.Wait() // 等待所有协程结束\n",
    "\tfmt.Printf(\"最终结果 counter=%d（预期值=%d）\\n\", counter, workers)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "最终结果 counter=1000（预期值=1000）\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "var (\n",
    "\tlock    sync.Mutex // 互斥锁：同一时刻只有一个协程进入临界区\n",
    "\tcounter int32       // 共享计数器\n",
    ")\n",
    "\n",
    "func safeIncrement(wg *sync.WaitGroup) {\n",
    "\tdefer wg.Done()\n",
    "\t\n",
    "\n",
    "\tlock.Lock()   // 加写锁，确保下面的 counter++ 在临界区串行执行\n",
    "\tcounter++     // 临界区：对共享变量的写操作\n",
    "\tdefer lock.Unlock() // 释放写锁，唤醒其他等待协程\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst workers = 1000\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo safeIncrement(&wg) // 并发启动 1000 个协程递增 counter\n",
    "\t}\n",
    "\n",
    "\twg.Wait() // 等待所有协程结束\n",
    "\tfmt.Printf(\"最终结果 counter=%d（预期值=%d）\\n\", counter, workers)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### sync.RWMutex 读锁可重入，写锁不可重入"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `RLock` 支持同一协程重复进入；适合递归读取或多层逻辑共享同一读锁。\n",
    "- `Lock` 不可重入，示例最终会触发 `fatal error`（非 panic，无法 recover）。\n",
    "- 实际业务中，写锁的递归或重复获取需格外谨慎，避免死锁。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 读锁可重入示例 ===\n",
      "read level: 3\n",
      "read level: 2\n",
      "read level: 1\n",
      "read level: 0\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var mu sync.RWMutex // 读写锁：同时提供读锁/写锁能力\n",
    "\n",
    "func readRecursive(n int) {\n",
    "\tmu.RLock()         // 获取读锁（可重入）\n",
    "\tdefer mu.RUnlock() // 函数退出时释放读锁\n",
    "\tfmt.Println(\"read level:\", n)\n",
    "\n",
    "\tif n > 0 {\n",
    "\t\treadRecursive(n - 1) // 递归调用：同一协程再次获取读锁不会阻塞\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tfmt.Println(\"=== 读锁可重入示例 ===\")\n",
    "\treadRecursive(3)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 写锁不可重入示例（运行后将触发 fatal error）===\n",
      "write level: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "fatal error: all goroutines are asleep - deadlock!\n",
      "\n",
      "goroutine 1 [select (no cases)]:\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [19] Line 31 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:156 +0x57\n",
      "\n",
      "goroutine 6 [sync.Mutex.Lock]:\n",
      "internal/sync.runtime_SemacquireMutex(0xc000088140?, 0x0?, 0x0?)\n",
      "\t/usr/local/go/src/runtime/sema.go:95 +0x25\n",
      "internal/sync.(*Mutex).lockSlow(0x57f2a0)\n",
      "\t/usr/local/go/src/internal/sync/mutex.go:149 +0x15d\n",
      "internal/sync.(*Mutex).Lock(...)\n",
      "\t/usr/local/go/src/internal/sync/mutex.go:70\n",
      "sync.(*Mutex).Lock(...)\n",
      "\t/usr/local/go/src/sync/mutex.go:46\n",
      "sync.(*RWMutex).Lock(0x0?)\n",
      "\t/usr/local/go/src/sync/rwmutex.go:150 +0x31\n",
      "main.writeRecursive(0x0)\n",
      "\t \u001b[7m[[ Cell Line 0 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:118 +0x2f\n",
      "main.writeRecursive(0x1)\n",
      "\t \u001b[7m[[ Cell Line 0 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:124 +0xcd\n",
      "main.main.func1()\n",
      "\t \u001b[7m[[ Cell [19] Line 27 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:152 +0x25\n",
      "created by main.main in goroutine 1\n",
      "\t \u001b[7m[[ Cell [19] Line 24 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:149 +0x52\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var mu sync.RWMutex // 读写锁：同时提供读锁/写锁能力\n",
    "\n",
    "func writeRecursive(n int) {\n",
    "\tmu.Lock()         // 获取写锁（不可重入）\n",
    "\tdefer mu.Unlock() // 函数退出时释放写锁\n",
    "\tfmt.Println(\"write level:\", n)\n",
    "\n",
    "\tif n > 0 {\n",
    "\t\t// 这里会导致死锁：写锁不可重入，再次 Lock() 会永远等待\n",
    "\t\twriteRecursive(n - 1)\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tfmt.Println(\"=== 写锁不可重入示例（运行后将触发 fatal error）===\")\n",
    "\tgo func() {\n",
    "\t\t// 给主协程一点时间打印提示信息\n",
    "\t\ttime.Sleep(200 * time.Millisecond)\n",
    "\t\twriteRecursive(1) // 第二次 Lock() 会造成致命死锁\n",
    "\t}()\n",
    "\n",
    "\t// 阻塞等待 fatal error 发生\n",
    "\tselect {}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 读写互斥与写锁优先排队"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 一旦写锁进入等待队列，后续读锁也要等待（防止写锁饥饿）。\n",
    "- 日志可看到锁按照排队顺序依次获得：主 → 协程1 → 协程2 → 写锁（协程3、4）。\n",
    "- 排队机制确保公平性：先请求写锁的优先于后面的读锁，即使读锁只晚了 100ms。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "reader 2 sees data=0\n",
      "reader 0 sees data=0\n",
      "reader 1 sees data=0\n",
      "writer 1 updates data=42\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var sharedData int\n",
    "var guard sync.RWMutex\n",
    "\n",
    "func reader(id int) {\n",
    "\tguard.RLock()         // 申请读锁：允许多个读者并行\n",
    "\tdefer guard.RUnlock() // 函数返回前释放读锁\n",
    "\tfmt.Printf(\"reader %d sees data=%d\\n\", id, sharedData)\n",
    "\ttime.Sleep(200 * time.Millisecond) // 模拟读取耗时\n",
    "}\n",
    "\n",
    "func writer(id int, value int) {\n",
    "\tguard.Lock()         // 申请写锁：阻塞所有其他读写请求\n",
    "\tdefer guard.Unlock() // 函数返回前释放写锁\n",
    "\tsharedData = value\n",
    "\tfmt.Printf(\"writer %d updates data=%d\\n\", id, value)\n",
    "\ttime.Sleep(300 * time.Millisecond) // 模拟写入耗时\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\t// 启动多个读者\n",
    "\tfor i := 0; i < 3; i++ {\n",
    "\t\tgo reader(i)\n",
    "\t}\n",
    "\n",
    "\ttime.Sleep(50 * time.Millisecond) // 确保读者先运行\n",
    "\n",
    "\t// 写者尝试写入：必须等所有读者释放读锁\n",
    "\tgo writer(1, 42)\n",
    "\n",
    "\t// 等待所有输出完成\n",
    "\ttime.Sleep(2 * time.Second)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 读锁与写锁互斥演示"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 多个读者可同时读取共享数据。\n",
    "- 写者必须等所有读锁释放后才能进入；进入写锁期间新的读者也会被阻塞。\n",
    "- 读多写少场景下，`RWMutex` 能显著提升读并发。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "reader 2 sees data=0\n",
      "reader 0 sees data=0\n",
      "reader 1 sees data=0\n",
      "writer 1 updates data=42\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var sharedData int\n",
    "var guard sync.RWMutex\n",
    "\n",
    "func reader(id int) {\n",
    "\tguard.RLock()         // 申请读锁：允许多个读者并行\n",
    "\tdefer guard.RUnlock() // 函数返回前释放读锁\n",
    "\tfmt.Printf(\"reader %d sees data=%d\\n\", id, sharedData)\n",
    "\ttime.Sleep(200 * time.Millisecond) // 模拟读取耗时\n",
    "}\n",
    "\n",
    "func writer(id int, value int) {\n",
    "\tguard.Lock()         // 申请写锁：阻塞所有其他读写请求\n",
    "\tdefer guard.Unlock() // 函数返回前释放写锁\n",
    "\tsharedData = value\n",
    "\tfmt.Printf(\"writer %d updates data=%d\\n\", id, value)\n",
    "\ttime.Sleep(300 * time.Millisecond) // 模拟写入耗时\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\t// 启动多个读者\n",
    "\tfor i := 0; i < 3; i++ {\n",
    "\t\tgo reader(i)\n",
    "\t}\n",
    "\n",
    "\ttime.Sleep(50 * time.Millisecond) // 确保读者先运行\n",
    "\n",
    "\t// 写者尝试写入：必须等所有读者释放读锁\n",
    "\tgo writer(1, 42)\n",
    "\n",
    "\t// 等待所有输出完成\n",
    "\ttime.Sleep(2 * time.Second)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 在读操作占绝大多数、写操作非常少的场景下的性能比较"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `sync.Mutex` 无论读写都要进行互斥，导致所有协程排队；读写比越悬殊，浪费越明显。\n",
    "- `sync.RWMutex` 对读操作使用共享锁，可同时处理多个读者，仅在写操作时才独占锁，因此在读多写少时具有更好的并行度与吞吐性能。\n",
    "- 若写操作比例显著升高或写操作持锁时间较长，则 `sync.RWMutex` 的调度开销也会上升，此时未必优于 `sync.Mutex`，需要结合实际 workload 测试。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Demo with sync.Mutex:\n",
      "Final counter value: 400\n",
      "\n",
      "Demo with sync.RWMutex:\n",
      "Final counter value: 400\n"
     ]
    }
   ],
   "source": [
    "! cd ./91 && go run ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "goos: linux\n",
      "goarch: amd64\n",
      "pkg: example.com/testmutex/testmutex\n",
      "cpu: AMD Ryzen 7 7840HS with Radeon 780M Graphics   \n",
      "BenchmarkMutexReadMostly-16      \t41994848\t        31.11 ns/op\t       0 B/op\t       0 allocs/op\n",
      "BenchmarkRWMutexReadMostly-16    \t202407993\t         6.242 ns/op\t       0 B/op\t       0 allocs/op\n",
      "PASS\n",
      "ok  \texample.com/testmutex/testmutex\t3.211s\n"
     ]
    }
   ],
   "source": [
    "! cd ./91/testmutex && go test -bench=. -benchmem"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "goos: linux\n",
      "goarch: amd64\n",
      "pkg: example.com/testmutex/testmutex\n",
      "cpu: AMD Ryzen 7 7840HS with Radeon 780M Graphics   \n",
      "BenchmarkMutexReadMostly   \t323556420\t         3.681 ns/op\t       0 B/op\t       0 allocs/op\n",
      "BenchmarkRWMutexReadMostly \t337158658\t         3.548 ns/op\t       0 B/op\t       0 allocs/op\n",
      "PASS\n",
      "ok  \texample.com/testmutex/testmutex\t3.130s\n"
     ]
    }
   ],
   "source": [
    "! cd ./91/testmutex && go test -bench=. -benchmem -cpu=1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "goos: linux\n",
      "goarch: amd64\n",
      "pkg: example.com/testmutex/testmutex\n",
      "cpu: AMD Ryzen 7 7840HS with Radeon 780M Graphics   \n",
      "BenchmarkMutexReadMostly-2     \t175361154\t         6.990 ns/op\t       0 B/op\t       0 allocs/op\n",
      "BenchmarkRWMutexReadMostly-2   \t273445446\t         4.375 ns/op\t       0 B/op\t       0 allocs/op\n",
      "PASS\n",
      "ok  \texample.com/testmutex/testmutex\t3.567s\n"
     ]
    }
   ],
   "source": [
    "! cd ./91/testmutex && go test -bench=. -benchmem -cpu=2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 92 如何并行修改结构体、切片、map"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 数据结构 / 操作场景  | 并发特性                                   | 推荐策略                                                     | 风险提示                                                     |\n",
    "| -------------------- | ------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |\n",
    "| 数组 (`[N]T`)        | 底层内存固定，允许多个协程并行写入不同索引 | 预先划分索引区间（如偶数位 vs 奇数位），避免同一元素被多 goroutine 修改 | 若修改同一索引，将出现数据竞争，结果不可预期                 |\n",
    "| 切片 (`[]T`)         | 引用底层数组，与数组同构                   | 仍需划分索引区间；若涉及 `append` 导致扩容，须序列化处理     | 扩容会分配新底层数组，可能造成数据错乱或 panic               |\n",
    "| 结构体 (`struct`)    | 多字段共享同一内存块                       | 按字段或责任分区（协程 A 写 `Name`，协程 B 写 `Age`），确保字段不冲突 | 同一字段并发写需加锁或使用原子操作                           |\n",
    "| 内置 map (`map[K]V`) | 默认非并发安全，写操作存在致命风险         | 1) 外层加互斥锁；2) 使用 `sync.Map`；3) 单线程序列化更新 + 通道传输 | 并发写会触发 `fatal error: concurrent map writes`，无法被 `recover` 捕获 |\n",
    "| `sync.Map`           | 面向场景优化的并发 map                     | `Store` / `Load` / `LoadOrStore` 等方法线程安全              | 适合读多写少或写入键不固定的场景；性能不一定优于手动加锁     |\n",
    "| 竞态检测             | `go run -race`                             | 在开发阶段检测数据竞争                                       | 正式环境不要开启（性能损耗大）                               |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**核心原则：**\n",
    "\n",
    "- “并行”意味着逻辑划分明确：每个 goroutine 负责互不重叠的内存区或字段。\n",
    "- 当无法避免共享写时，选择互斥锁 (`sync.Mutex` / `sync.RWMutex`) 或原子操作。\n",
    "- `map` 是特例：即使写不同的 key 也可能崩溃，需额外保护。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 数组按索引分区并发写入（无锁方案）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 两个 goroutine 修改的索引不重叠，因此不会出现数据竞争。\n",
    "- 若协程之间没有清晰的索引分工（如同时写 `arr[3]`），就会出现脏写，结果不确定。\n",
    "- 对切片同样适用（切片底层是数组），只要避免多个协程修改同一位置即可。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "最终数组状态: [0 1 0 1 0 1 0 1 0 1]\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar (\n",
    "\t\tarr = [10]int{}    // 固定长度数组，初始值全为 0\n",
    "\t\twg  sync.WaitGroup // 用于等待两个写协程完成\n",
    "\t)\n",
    "\n",
    "\twg.Add(2)\n",
    "\n",
    "\t// 协程 A：写偶数索引 (0,2,4,...)\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 0; i < len(arr); i += 2 {\n",
    "\t\t\tarr[i] = 0 // 按约定写入，永远不会与协程 B 冲突\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 协程 B：写奇数索引 (1,3,5,...)\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 1; i < len(arr); i += 2 {\n",
    "\t\t\tarr[i] = 1 // 与协程 A 写入的索引互不重叠\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"最终数组状态:\", arr)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 切片分段并行填充（注意 index 范围）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 切片与数组共用底层数组，只要划分好 index 区间，多个协程即可并行写入，提高吞吐。\n",
    "- 若在协程内执行 `append` 导致扩容，会生成新的底层数组，可能破坏其他协程的写入；因此在并行写入时禁止扩容。\n",
    "- 针对只读场景（多个协程读同一切片），无需加锁；只要不修改内容即可。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "切片填充完成: [10 11 12 13 24 25 26 27 38 39 40 41]\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tconst size = 12\n",
    "\tdata := make([]int, size) // 切片底层数组长度为 12\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(3)\n",
    "\n",
    "\t// 子任务 1：处理 [0,4)\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 0; i < 4; i++ {\n",
    "\t\t\tdata[i] = 10 + i\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 子任务 2：处理 [4,8)\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 4; i < 8; i++ {\n",
    "\t\t\tdata[i] = 20 + i\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 子任务 3：处理 [8,12)\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 8; i < 12; i++ {\n",
    "\t\t\tdata[i] = 30 + i\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"切片填充完成:\", data)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 结构体字段分工并发写（避免字段冲突）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 协程之间按字段分工，互不干扰。\n",
    "- 若多个协程同时递增 `Age` (`stu.Age++`)，照样会出现数据竞争，因为 `Age` 字段的写操作重叠。应使用锁或原子操作保护。\n",
    "- 此示例说明“并行修改结构体”并非完全禁止，而是要求协程之间的写入范围明确、无交集。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "结构体最终状态: Name=Fred Age=20\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "type Student struct {\n",
    "\tName string\n",
    "\tAge  int32\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tvar (\n",
    "\t\tstu Student\n",
    "\t\twg  sync.WaitGroup\n",
    "\t)\n",
    "\n",
    "\twg.Add(2)\n",
    "\n",
    "\t// 协程 A：只负责写 Name\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tstu.Name = \"Fred\"\n",
    "\t}()\n",
    "\n",
    "\t// 协程 B：只负责写 Age\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tstu.Age = 20\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Printf(\"结构体最终状态: Name=%s Age=%d\\n\", stu.Name, stu.Age)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 原生 map 并发写导致 fatal error"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 即便两个协程写不同的 key，运行时仍可能判定为并发冲突并崩溃。\n",
    "- `fatal error` 无法被 `recover()` 捕获，线上服务会直接异常退出。\n",
    "- 生产环境中绝不能直接并发写原生 map。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "fatal error: concurrent map writes\n",
      "\n",
      "goroutine 6 [running]:\n",
      "internal/runtime/maps.fatal({0x4b7f05?, 0x0?})\n",
      "\t/usr/local/go/src/runtime/panic.go:1058 +0x18\n",
      "main.main.func1()\n",
      "\t \u001b[7m[[ Cell [32] Line 20 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:151 +0x65\n",
      "created by main.main in goroutine 1\n",
      "\t \u001b[7m[[ Cell [32] Line 17 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:148 +0x93\n",
      "\n",
      "goroutine 1 [sync.WaitGroup.Wait]:\n",
      "sync.runtime_SemacquireWaitGroup(0xc00000e030?)\n",
      "\t/usr/local/go/src/runtime/sema.go:110 +0x25\n",
      "sync.(*WaitGroup).Wait(0xc00009af40?)\n",
      "\t/usr/local/go/src/sync/waitgroup.go:118 +0x48\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [32] Line 32 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:163 +0xef\n",
      "\n",
      "goroutine 7 [runnable]:\n",
      "internal/runtime/maps.typedmemmove(0x49c8e0?, 0xc0000cb500?, 0xc0000c2520?)\n",
      "\t/usr/local/go/src/runtime/mbarrier.go:235 +0x2d\n",
      "internal/runtime/maps.(*table).uncheckedPutSlot(0xc00007e120, 0x4a2580, 0xffffffffffffffff?, 0xc00007e001?, 0xc0000c2528)\n",
      "\t/usr/local/go/src/internal/runtime/maps/table.go:397 +0xd6\n",
      "internal/runtime/maps.(*table).split(0xc00007e0e0, 0x4a2580, 0xc0000a00f0)\n",
      "\t/usr/local/go/src/internal/runtime/maps/table.go:1079 +0x1b1\n",
      "internal/runtime/maps.(*table).rehash(0x0?, 0x0?, 0x0?)\n",
      "\t/usr/local/go/src/internal/runtime/maps/table.go:1031 +0x30\n",
      "main.main.func2()\n",
      "\t \u001b[7m[[ Cell [32] Line 28 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:159 +0x66\n",
      "created by main.main in goroutine 1\n",
      "\t \u001b[7m[[ Cell [32] Line 25 ]]\u001b[0m /tmp/gonb_9808e9c4/main.go:156 +0xe5\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "// 多运行几次 go run .，很可能看到程序直接崩溃并打印 fatal error: concurrent map writes。这是 Go 运行时在检测到 map 写操作竞争后主动终止程序，避免数据破坏。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tmp := make(map[int]bool, 10)\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(2)\n",
    "\n",
    "\t// 协程 A：写偶数 key\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 0; i < 10_000; i += 2 {\n",
    "\t\t\tmp[i] = true\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 协程 B：写奇数 key\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfor i := 1; i < 10_000; i += 2 {\n",
    "\t\t\tmp[i] = false\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"map[1] =\", mp[1])\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 使用 sync.Map 并发读写 map"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `sync.Map` 支持并发读写，`Store` / `Load` 方法无需额外加锁。\n",
    "- 多个协程同时写同一个 key 时，最终值取决于谁最后写入。\n",
    "- 与原生 map 不同，即使竞争写同一个 key，程序也不会崩溃。\n",
    "- `Load` 返回 `(value, bool)`，使用前必须检查 `bool`（是否存在）。\n",
    "- 适合键数量动态、读多写少、临时缓存等场景。如果 key 集合固定且写频繁，选择 `map + sync.RWMutex` 可能更高效。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "读取到的最终值: v1\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar (\n",
    "\t\tm  sync.Map   // 并发安全 map\n",
    "\t\twg sync.WaitGroup\n",
    "\t)\n",
    "\n",
    "\twg.Add(2)\n",
    "\n",
    "\t// 协程 A：写入 key = \"k1\"\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tm.Store(\"k1\", \"v1\") // Store 是并发安全的写操作\n",
    "\t}()\n",
    "\n",
    "\t// 协程 B：覆盖同一 key\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tm.Store(\"k1\", \"v2\") // 可能覆盖 v1，两个协程不会引起崩溃\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\n",
    "\t// 读取时需检查是否存在\n",
    "\tif val, ok := m.Load(\"k1\"); ok {\n",
    "\t\tfmt.Println(\"读取到的最终值:\", val)\n",
    "\t} else {\n",
    "\t\tfmt.Println(\"key 不存在\")\n",
    "\t}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 93 读写锁和泛型的综合练习--并发安全map"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "reader-1 observed writer-0-0\n",
      "reader-2 observed writer-0-0\n",
      "reader-3 observed writer-0-0\n",
      "reader-0 observed writer-0-0\n",
      "reader-2 observed writer-0-0\n",
      "reader-1 observed writer-0-0\n",
      "reader-3 observed writer-0-0\n",
      "reader-0 observed writer-0-0\n",
      "reader-1 observed writer-0-0\n",
      "reader-3 observed writer-0-0\n",
      "reader-3 observed writer-0-0\n",
      "reader-0 observed writer-0-0\n",
      "reader-0 observed writer-0-0\n",
      "reader-2 observed writer-0-0\n",
      "reader-1 observed writer-0-0\n",
      "reader-2 observed writer-0-0\n",
      "sample entry => writer-0-3: 3\n",
      "sample entry => writer-1-386: 386\n",
      "sample entry => writer-1-880: 880\n",
      "sample entry => writer-1-988: 988\n",
      "sample entry => writer-0-183: 183\n",
      "total entries written: 2000\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "// ConcurrentMap 是一个基于 Go 泛型的并发安全 map。\n",
    "// K：键类型，必须实现 comparable；V：值类型，任意。\n",
    "// 底层通过 sync.RWMutex 解决原生 map 并发读写不安全的问题。\n",
    "type ConcurrentMap[K comparable, V any] struct {\n",
    "\tdata map[K]V     // 原生 map 作为实际存储容器\n",
    "\tmu   sync.RWMutex // 读写锁：写操作独占、读操作共享\n",
    "}\n",
    "\n",
    "// NewConcurrentMap 创建一个新的并发安全 map。\n",
    "// cap 用于预分配底层 map 的容量，提升高并发下的性能表现。\n",
    "func NewConcurrentMap[K comparable, V any](cap int) *ConcurrentMap[K, V] {\n",
    "\treturn &ConcurrentMap[K, V]{\n",
    "\t\tdata: make(map[K]V, cap),\n",
    "\t}\n",
    "}\n",
    "\n",
    "// Store 写入或更新键值对：写操作必须持有写锁，防止数据竞争。\n",
    "func (cm *ConcurrentMap[K, V]) Store(key K, value V) {\n",
    "\tcm.mu.Lock()         // 写锁是独占锁，其它读/写都会被阻塞\n",
    "\tdefer cm.mu.Unlock() // defer 确保异常或提前返回时也能释放锁\n",
    "\tcm.data[key] = value\n",
    "}\n",
    "\n",
    "// Load 读取键值对：读操作使用读锁，支持并发读取，写锁优先级更高。\n",
    "func (cm *ConcurrentMap[K, V]) Load(key K) (value V, exists bool) {\n",
    "\tcm.mu.RLock()         // 读锁允许多个读者同时进入临界区\n",
    "\tdefer cm.mu.RUnlock() // 退出临界区时释放读锁\n",
    "\tvalue, exists = cm.data[key]\n",
    "\treturn\n",
    "}\n",
    "\n",
    "// Range 以读锁方式遍历 map，每个元素交给回调函数 f 处理。\n",
    "// 若 f 返回非 nil 错误，则提前结束遍历；否则遍历整个 map。\n",
    "func (cm *ConcurrentMap[K, V]) Range(f func(key K, value V) error) {\n",
    "\tcm.mu.RLock()\n",
    "\tdefer cm.mu.RUnlock()\n",
    "\tfor k, v := range cm.data {\n",
    "\t\tif err := f(k, v); err != nil {\n",
    "\t\t\tbreak // 提前结束遍历，常用场景：找到目标即可退出\n",
    "\t\t}\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tcm := NewConcurrentMap[string, int](100)\n",
    "\n",
    "\tconst (\n",
    "\t\treaders = 5\n",
    "\t\twriters = 2\n",
    "\t\tloops   = 1000\n",
    "\t)\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\n",
    "\t// 并发写入：演示写锁的独占特性\n",
    "\tfor i := 0; i < writers; i++ {\n",
    "\t\twg.Add(1)\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tfor j := 0; j < loops; j++ {\n",
    "\t\t\t\tkey := fmt.Sprintf(\"writer-%d-%d\", id, j)\n",
    "\t\t\t\tcm.Store(key, j)\n",
    "\t\t\t}\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\t// 并发读取：演示读锁允许多个 goroutine 同时读取\n",
    "\tfor i := 0; i < readers; i++ {\n",
    "\t\twg.Add(1)\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tfor j := 0; j < loops; j++ {\n",
    "\t\t\t\tif _, exists := cm.Load(\"writer-0-0\"); exists && j%250 == 0 {\n",
    "\t\t\t\t\tfmt.Printf(\"reader-%d observed writer-0-0\\n\", id)\n",
    "\t\t\t\t}\n",
    "\t\t\t}\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\n",
    "\t// 使用 Range 遍历部分结果\n",
    "\tcount := 0\n",
    "\tcm.Range(func(key string, value int) error {\n",
    "\t\tif count < 5 {\n",
    "\t\t\tfmt.Printf(\"sample entry => %s: %d\\n\", key, value)\n",
    "\t\t}\n",
    "\t\tcount++\n",
    "\t\treturn nil\n",
    "\t})\n",
    "\tfmt.Printf(\"total entries written: %d\\n\", count)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== RUN   TestConcurrentMap\n",
      "=== PAUSE TestConcurrentMap\n",
      "=== RUN   TestConcurrentSafety\n",
      "=== PAUSE TestConcurrentSafety\n",
      "=== RUN   TestRange\n",
      "=== PAUSE TestRange\n",
      "=== CONT  TestConcurrentMap\n",
      "--- PASS: TestConcurrentMap (0.00s)\n",
      "=== CONT  TestConcurrentSafety\n",
      "=== CONT  TestRange\n",
      "--- PASS: TestRange (0.00s)\n",
      "--- PASS: TestConcurrentSafety (0.00s)\n",
      "PASS\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"testing\"\n",
    ")\n",
    "\n",
    "// TestConcurrentMap 基础读写功能验证：支持不同键、值类型，并正确返回存在/不存在结果。\n",
    "func TestConcurrentMap(t *testing.T) {\n",
    "\tt.Parallel()\n",
    "\n",
    "\tcm := NewConcurrentMap[string, int](16)\n",
    "\tcm.Store(\"张三\", 18)\n",
    "\n",
    "\tif v, ok := cm.Load(\"张三\"); !ok {\n",
    "\t\tt.Fatalf(\"expected 张三 to exist\")\n",
    "\t} else if v != 18 {\n",
    "\t\tt.Fatalf(\"expected 18, got %d\", v)\n",
    "\t}\n",
    "\n",
    "\tif _, ok := cm.Load(\"李四\"); ok {\n",
    "\t\tt.Fatalf(\"expected 李四 not to exist\")\n",
    "\t}\n",
    "\n",
    "\t// 验证泛型能力：键为 int，值为任意结构体\n",
    "\ttype user struct {\n",
    "\t\tID   int\n",
    "\t\tName string\n",
    "\t}\n",
    "\tusers := NewConcurrentMap[int, user](4)\n",
    "\tusers.Store(1, user{ID: 1, Name: \"Alice\"})\n",
    "\tif got, ok := users.Load(1); !ok || got.Name != \"Alice\" {\n",
    "\t\tt.Fatalf(\"generic map failed, got=%+v, ok=%v\", got, ok)\n",
    "\t}\n",
    "}\n",
    "\n",
    "// TestRange 遍历函数测试：确保在回调中能访问全部元素，并可提前终止。\n",
    "func TestRange(t *testing.T) {\n",
    "\tt.Parallel()\n",
    "\n",
    "\tcm := NewConcurrentMap[int, bool](4)\n",
    "\tfor i := 0; i < 10; i++ {\n",
    "\t\tcm.Store(i, true)\n",
    "\t}\n",
    "\n",
    "\tcount := 0\n",
    "\tcm.Range(func(key int, value bool) error {\n",
    "\t\tcount++\n",
    "\t\tif key == 5 {\n",
    "\t\t\treturn fmt.Errorf(\"stop\") // 返回错误，提前终止\n",
    "\t\t}\n",
    "\t\treturn nil\n",
    "\t})\n",
    "\n",
    "\tif count == 10 {\n",
    "\t\tt.Fatalf(\"expected early break, but iterated all entries\")\n",
    "\t}\n",
    "\tif count == 0 {\n",
    "\t\tt.Fatalf(\"range callback not invoked\")\n",
    "\t}\n",
    "}\n",
    "\n",
    "// TestConcurrentSafety 并发读写压力测试：多个 goroutine 写入同一 map，验证无数据竞争。\n",
    "func TestConcurrentSafety(t *testing.T) {\n",
    "\tt.Parallel()\n",
    "\n",
    "\tconst (\n",
    "\t\tworkers    = 10\n",
    "\t\twriteLoops = 1000\n",
    "\t)\n",
    "\n",
    "\tcm := NewConcurrentMap[int, bool](workers * writeLoops)\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tfor j := 0; j < writeLoops; j++ {\n",
    "\t\t\t\tcm.Store(j, true)\n",
    "\t\t\t}\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\twg.Wait()\n",
    "\n",
    "\t// spot check\n",
    "\tfor _, key := range []int{0, writeLoops / 2, writeLoops - 1} {\n",
    "\t\tif v, ok := cm.Load(key); !ok || !v {\n",
    "\t\t\tt.Fatalf(\"expected key %d to exist with value true\", key)\n",
    "\t\t}\n",
    "\t}\n",
    "}\n",
    "\n",
    "%test"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 94 recover与协程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题       | 核心概念                         | 关键要点                                     | 常见误区 / 风险                     | 应用建议                                     |\n",
    "| ---------- | -------------------------------- | -------------------------------------------- | ----------------------------------- | -------------------------------------------- |\n",
    "| `panic`    | Go 的运行时错误信号              | 触发后立即展开栈，沿调用链向上查找 `defer`   | 未捕获 `panic` 会导致整个进程崩溃   | 仅在真正的不可恢复错误场景使用               |\n",
    "| `recover`  | 使当前协程从 `panic` 中恢复      | 必须**搭配 `defer`**、**放在匿名函数**内调用 | 写在父协程无法捕获子协程的 `panic`  | 每个协程/请求入口单独封装 `recover`          |\n",
    "| 协程隔离   | 每个 `goroutine` 拥有独立调用栈  | `recover` 只能捕获**当前协程**内的 `panic`   | 在父协程写 `recover` 无效           | 封装统一的 `safeGo(func())` 工具             |\n",
    "| 微服务接口 | 每个请求通常独立运行在新的协程   | 任一请求触发 `panic` 将直接杀死进程          | 忽视参数校验引发越界、除零等异常    | 在每个 handler 顶层添加 `recover` 并记录错误 |\n",
    "| 错误治理   | 捕获后需日志/告警/回包           | 简单 `recover()` 而不处理等于吞错            | `fatal error` 不可被 `recover` 捕获 | 捕获后务必记录上下文并返回友好错误           |\n",
    "| 调试技巧   | `go test -run`、`panic` 堆栈分析 | 单元测试中验证异常路径                       | 误将 `recover` 放在循环外导致失效   | 封装 reusable 中间件函数确保生效             |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 每个协程独立 recover"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `recover` 必须写在协程内部；本例中每个 handler 都自带 `defer-recover`。\n",
    "- 即便出现 `panic`，服务仍能继续运行。\n",
    "- 演示了数组越界与除零两种典型异常。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025/09/30 04:23:00 [safeSetHandler] panic 已捕获: runtime error: integer divide by zero (index=0, div=0)\n",
      "2025/09/30 04:23:00 [safeGetHandler] panic 已捕获: runtime error: index out of range [10] with length 7 (index=10)\n",
      "2025/09/30 04:23:01 main 正常退出，进程未崩溃\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"log\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "var table = []int{1, 2, 3, 4, 5, 6, 7}\n",
    "\n",
    "// safeGetHandler 在当前协程内捕获可能的越界 panic\n",
    "func safeGetHandler(index int) {\n",
    "\tdefer func() {\n",
    "\t\tif err := recover(); err != nil {\n",
    "\t\t\tlog.Printf(\"[safeGetHandler] panic 已捕获: %v (index=%d)\\n\", err, index)\n",
    "\t\t}\n",
    "\t}()\n",
    "\tfmt.Printf(\"table[%d] = %d\\n\", index, table[index]) // 可能越界\n",
    "}\n",
    "\n",
    "// safeSetHandler 在当前协程内捕获类型转换和除零 panic\n",
    "func safeSetHandler(index int, div string) {\n",
    "\tdefer func() {\n",
    "\t\tif err := recover(); err != nil {\n",
    "\t\t\tlog.Printf(\"[safeSetHandler] panic 已捕获: %v (index=%d, div=%s)\\n\", err, index, div)\n",
    "\t\t}\n",
    "\t}()\n",
    "\tvalue := parseAndDivide(div)\n",
    "\ttable[index] = value // 只有在除零未发生时才写入\n",
    "\tfmt.Printf(\"table[%d] 更新为 %d\\n\", index, value)\n",
    "}\n",
    "\n",
    "// parseAndDivide 将字符串转换为整数并执行整除\n",
    "func parseAndDivide(div string) int {\n",
    "\t// 故意忽略 error，触发 panic（示例用）\n",
    "\tn := mustAtoi(div)\n",
    "\treturn 10 / n // 若 n 为 0，会 panic\n",
    "}\n",
    "\n",
    "// mustAtoi 仅作示例：转换失败会 panic\n",
    "func mustAtoi(s string) int {\n",
    "\tvar n int\n",
    "\tfmt.Sscan(s, &n) // 写入失败时 n 保持 0，除法环节触发 panic\n",
    "\treturn n\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tgo safeGetHandler(10) // 越界 -> panic -> recover\n",
    "\tgo safeSetHandler(0, \"0\") // 除零 -> panic -> recover\n",
    "\n",
    "\ttime.Sleep(500 * time.Millisecond) // 等待子协程输出\n",
    "\tlog.Println(\"main 正常退出，进程未崩溃\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 父协程的 recover 无法捕获子协程的 panic"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 证明了“`recover` 只能捕获本协程 `panic`”的特性。\n",
    "- 若想捕获子协程 `panic`，必须在子协程内部封装 `defer-recover`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025/09/30 04:25:54 [child] 即将发生 panic\n",
      "panic: 子协程 panic\n",
      "\n",
      "goroutine 6 [running]:\n",
      "main.main.func2()\n",
      "\t \u001b[7m[[ Cell [4] Line 22 ]]\u001b[0m /tmp/gonb_3f9c0917/main.go:190 +0x86\n",
      "created by main.main in goroutine 1\n",
      "\t \u001b[7m[[ Cell [4] Line 20 ]]\u001b[0m /tmp/gonb_3f9c0917/main.go:188 +0x3b\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "// 即使在 main 中写了 recover，程序仍会崩溃，且 defer 不会捕获到子协程的异常。\n",
    "// 控制台输出会看到 panic: 子协程 panic，随后程序退出。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"log\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tdefer func() {\n",
    "\t\tif err := recover(); err != nil {\n",
    "\t\t\tlog.Println(\"[main] 捕获 panic：\", err)\n",
    "\t\t} else {\n",
    "\t\t\tlog.Println(\"[main] 没有 panic 被捕获\")\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\tgo func() {\n",
    "\t\tlog.Println(\"[child] 即将发生 panic\")\n",
    "\t\tpanic(\"子协程 panic\")\n",
    "\t}()\n",
    "\n",
    "\ttime.Sleep(500 * time.Millisecond)\n",
    "\tlog.Println(\"[main] 仍然会因子协程 panic 而崩溃\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 模拟微服务接口，用 recover 保护请求"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `safeGo` 统一封装了 `defer-recover`，模拟微服务框架为每个请求启动协程的行为。\n",
    "- 每个 handler 内部仍保留 `recover`，避免开发者遗漏。\n",
    "- 即使出现数组越界、字符串转数字失败、除零等错误，都不会影响其他请求的执行。\n",
    "- 日志中包含错误上下文，方便排查"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025/09/30 04:26:41 [setHandler] 请求(非法写-除零)发生异常: runtime error: integer divide by zero\n",
      "2025/09/30 04:26:41 [getHandler] 请求(越界读)发生异常: runtime error: index out of range [10] with length 7\n",
      "2025/09/30 04:26:41 [setHandler] 请求(非法写-非数字)发生异常: 参数 \"abc\" 不是合法整数\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[合法读] table[2] = 3\n",
      "[合法写] table[1] 更新为 2\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025/09/30 04:26:42 所有请求已完成，服务仍保持运行状态\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"log\"\n",
    "\t\"strconv\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "type request struct {\n",
    "\tname  string\n",
    "\tindex int\n",
    "\tdiv   string\n",
    "}\n",
    "\n",
    "// safeGo 封装协程入口，统一捕获 panic，避免每个调用点手写 recover\n",
    "func safeGo(fn func()) {\n",
    "\tgo func() {\n",
    "\t\tdefer func() {\n",
    "\t\t\tif err := recover(); err != nil {\n",
    "\t\t\t\tlog.Printf(\"[safeGo] panic: %v\\n\", err)\n",
    "\t\t\t}\n",
    "\t\t}()\n",
    "\t\tfn()\n",
    "\t}()\n",
    "}\n",
    "\n",
    "// getHandler 微服务读接口：返回数组元素\n",
    "func getHandler(req request) {\n",
    "\tdefer func() {\n",
    "\t\tif err := recover(); err != nil {\n",
    "\t\t\tlog.Printf(\"[getHandler] 请求(%s)发生异常: %v\\n\", req.name, err)\n",
    "\t\t}\n",
    "\t}()\n",
    "\tfmt.Printf(\"[%s] table[%d] = %d\\n\", req.name, req.index, table[req.index])\n",
    "}\n",
    "\n",
    "// setHandler 微服务写接口：解析并更新数组元素\n",
    "func setHandler(req request) {\n",
    "\tdefer func() {\n",
    "\t\tif err := recover(); err != nil {\n",
    "\t\t\tlog.Printf(\"[setHandler] 请求(%s)发生异常: %v\\n\", req.name, err)\n",
    "\t\t}\n",
    "\t}()\n",
    "\tvalue := parseDiv(req.div)\n",
    "\ttable[req.index] = value\n",
    "\tfmt.Printf(\"[%s] table[%d] 更新为 %d\\n\", req.name, req.index, value)\n",
    "}\n",
    "\n",
    "var table = []int{1, 2, 3, 4, 5, 6, 7}\n",
    "\n",
    "func parseDiv(s string) int {\n",
    "\tn, err := strconv.Atoi(s)\n",
    "\tif err != nil {\n",
    "\t\tpanic(fmt.Sprintf(\"参数 %q 不是合法整数\", s))\n",
    "\t}\n",
    "\treturn 10 / n // n 为 0 时 panic\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\t// 模拟并发请求\n",
    "\trequests := []request{\n",
    "\t\t{name: \"合法读\", index: 2},\n",
    "\t\t{name: \"越界读\", index: 10},\n",
    "\t\t{name: \"合法写\", index: 1, div: \"5\"},\n",
    "\t\t{name: \"非法写-非数字\", index: 0, div: \"abc\"},\n",
    "\t\t{name: \"非法写-除零\", index: 2, div: \"0\"},\n",
    "\t}\n",
    "\n",
    "\tfor _, req := range requests {\n",
    "\t\tswitch {\n",
    "\t\tcase req.div == \"\":\n",
    "\t\t\tsafeGo(func(r request) func() { return func() { getHandler(r) } }(req))\n",
    "\t\tdefault:\n",
    "\t\t\tsafeGo(func(r request) func() { return func() { setHandler(r) } }(req))\n",
    "\t\t}\n",
    "\t}\n",
    "\n",
    "\ttime.Sleep(1 * time.Second)\n",
    "\tlog.Println(\"所有请求已完成，服务仍保持运行状态\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 95 channel的阻塞与遍历"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题         | 核心机制                                 | 关键 API / 行为                              | 常见坑点                                   | 实战建议                                      |\n",
    "| ------------ | ---------------------------------------- | -------------------------------------------- | ------------------------------------------ | --------------------------------------------- |\n",
    "| Channel 本质 | Goroutine 之间共享数据的安全队列         | `make(chan T, cap)` 创建（`cap=0` 为无缓冲） | 把 Channel 当普通队列使用但忽略阻塞特性    | 读写都应配对，必要时配合 `select` / `context` |\n",
    "| 阻塞规则     | 空通道读阻塞；满通道写阻塞               | `<-ch`、`ch<-`                               | 单协程里既读又写易造成自锁                 | 设计通信图时预估生产者/消费者速度             |\n",
    "| 缓冲行为     | 缓冲区允许“先写后读”                     | 带缓冲通道可提前写入 `cap` 次                | 误以为缓冲通道不会阻塞                     | 根据峰值和速率合理设置 `cap`                  |\n",
    "| 解除阻塞     | 需要其他 Goroutine 协助                  | 读阻塞需写入，写阻塞需读取                   | 忽视协程启动时序导致死锁                   | 保证读写协程都已启动且存活                    |\n",
    "| 关闭通道     | `close(ch)` 表示不会再写入               | `<-ch` 返回 `(零值, false)`                  | 向已关闭通道写会 panic；重复关闭也会 panic | 建立“谁负责关闭”的约定                        |\n",
    "| 遍历模式     | `for v := range ch` 消费所有元素直到关闭 | 需先 `close(ch)` 才能退出循环                | 未关闭通道导致遍历协程永久阻塞             | 生产者写完后关闭；或使用额外信号通知结束      |\n",
    "| FIFO 保证    | Channel 是先进先出队列                   | 写入顺序等于读取顺序                         | 不要用通道“临时调整”顺序                   | 若需自定义顺序，可考虑 `slice + Mutex`        |\n",
    "| Panic 场景   | 写已关闭通道、重复关闭                   | `panic: send on closed channel`              | 错误地把多协程的关闭职责混在一起           | 约定只有生产者关闭；或用 `sync.Once`          |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 演示写阻塞与解阻塞时序"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- channel 满时写操作阻塞，直到有元素被读取。\n",
    "- FIFO 行为保证 `take` 顺序与 `send` 顺序一致。\n",
    "- 设计时需保证有协程及时消费，否则写端可能永远阻塞。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "send 1\n",
      "send 2\n",
      "send 3\n",
      "take 1\n",
      "take 2\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tch := make(chan int, 2) // 缓冲为 2：允许先写入两个元素不阻塞\n",
    "\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(time.Second) // 让主协程先写入并阻塞\n",
    "\t\tif v, ok := <-ch; ok {\n",
    "\t\t\tfmt.Println(\"take\", v)\n",
    "\t\t}\n",
    "\t\ttime.Sleep(time.Second)\n",
    "\t\tif v, ok := <-ch; ok {\n",
    "\t\t\tfmt.Println(\"take\", v)\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\tch <- 1\n",
    "\tfmt.Println(\"send 1\") // 立即写入\n",
    "\tch <- 2\n",
    "\tfmt.Println(\"send 2\") // 仍未满，立即写入\n",
    "\n",
    "\t// 第三次写入时缓冲区已满，只能阻塞等待消费者读取\n",
    "\tch <- 3\n",
    "\tfmt.Println(\"send 3\") // 只有在第一个元素被读取后才打印\n",
    "\n",
    "\ttime.Sleep(2 * time.Second) // 等子协程输出完毕\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 关闭通道后的读取语义"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 关闭通道后且通道中没有待取的数据, 读取会立即返回 `(零值, false)`，无需等待。\n",
    "- 应通过双返回值判断是否读取到有效数据。\n",
    "- 向已关闭通道写入会 `panic`，必须由生产者负责关闭以避免误用。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "第一次读取: v=1 ok=true\n",
      "第二次读取: v=0 ok=false\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tch := make(chan int, 2)\n",
    "\tch <- 1\n",
    "\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(3 * time.Second)\n",
    "\t\tch <- 2\n",
    "\t}()\n",
    "\n",
    "\t// 关闭通道：后续禁止写入，但已写入元素仍可被读取\n",
    "\tclose(ch)\n",
    "\n",
    "\tv, ok := <-ch\n",
    "\tfmt.Printf(\"第一次读取: v=%d ok=%v\\n\", v, ok)\n",
    "\n",
    "\tv, ok = <-ch\n",
    "\tfmt.Printf(\"第二次读取: v=%d ok=%v\\n\", v, ok) // 立即返回零值，ok=false\n",
    "\n",
    "\t// 再次写入将直接 panic\n",
    "\t// ch <- 3\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "第一次读取: v=1 ok=true\n",
      "第二次读取: v=0 ok=false\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "panic: send on closed channel\n",
      "\n",
      "goroutine 1 [running]:\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [3] Line 27 ]]\u001b[0m /tmp/gonb_2ce7dd61/main.go:28 +0x1bf\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tch := make(chan int, 2)\n",
    "\tch <- 1\n",
    "\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(3 * time.Second)\n",
    "\t\tch <- 2\n",
    "\t}()\n",
    "\n",
    "\t// 关闭通道：后续禁止写入，但已写入元素仍可被读取\n",
    "\tclose(ch)\n",
    "\n",
    "\tv, ok := <-ch\n",
    "\tfmt.Printf(\"第一次读取: v=%d ok=%v\\n\", v, ok)\n",
    "\n",
    "\tv, ok = <-ch\n",
    "\tfmt.Printf(\"第二次读取: v=%d ok=%v\\n\", v, ok) // 立即返回零值，ok=false\n",
    "\n",
    "\t// 再次写入将直接 panic\n",
    "\tch <- 3\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 遍历通道与安全退出"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `for v := range ch` 会消费元素，直到通道被关闭且元素取尽。\n",
    "- 若生产者不关闭通道，消费者会永久阻塞在 `range`，导致死锁。\n",
    "- 只有“数据流结束”时才关闭通道；若数据流持续（例如流式服务），则不关闭，由写端持续发送。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "recv 1\n",
      "recv 2\n",
      "for-range 结束，通道已清空并关闭\n",
      "manual recv 10\n",
      "manual recv 20\n",
      "读取完毕，退出循环\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import \"fmt\"\n",
    "\n",
    "func main() {\n",
    "\tch := make(chan int, 2)\n",
    "\tch <- 1\n",
    "\tch <- 2\n",
    "\tclose(ch) // 必须关闭，否则 for-range 无法得知数据结束\n",
    "\n",
    "\t// for-range 等价于持续读取直到 ok == false\n",
    "\tfor v := range ch {\n",
    "\t\tfmt.Println(\"recv\", v)\n",
    "\t}\n",
    "\tfmt.Println(\"for-range 结束，通道已清空并关闭\")\n",
    "\n",
    "\t// 等价写法：显示检查 ok\n",
    "\tch = make(chan int, 2)\n",
    "\tch <- 10\n",
    "\tch <- 20\n",
    "\tclose(ch)\n",
    "\n",
    "\tfor {\n",
    "\t\tv, ok := <-ch\n",
    "\t\tif !ok {\n",
    "\t\t\tfmt.Println(\"读取完毕，退出循环\")\n",
    "\t\t\tbreak\n",
    "\t\t}\n",
    "\t\tfmt.Println(\"manual recv\", v)\n",
    "\t}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 96 阻塞代码的5种方法及导致死锁的根本原因"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 阻塞方式         | 典型用法                              | 解除条件                              | 是否依赖外部协程 | 常见误区 / 风险                                |\n",
    "| ---------------- | ------------------------------------- | ------------------------------------- | ---------------- | ---------------------------------------------- |\n",
    "| `time.Sleep`     | `time.Sleep(duration)`                | 时间耗尽自动恢复                      | 否               | 误以为能触发死锁；实际上它会自行恢复           |\n",
    "| `sync.WaitGroup` | `wg.Add(n)` + `wg.Wait()`             | 所有子任务执行 `wg.Done()` 使计数归零 | 是               | 忘记 `Add` 或 `Done` 会导致永久等待            |\n",
    "| Channel 操作     | `<-ch` / `ch <- val` / `for range ch` | 对端写入/读取或关闭通道               | 是               | 从空 channel 读取或遍历未关闭 channel 易致死锁 |\n",
    "| `sync.Mutex`     | `mu.Lock()`                           | 持有锁的协程执行 `Unlock()`           | 是               | 同一协程二次加锁或忘记 `Unlock` 导致自锁       |\n",
    "| 空 `select`      | `select {}`                           | 无永远阻塞                            | 否               | 调试测试中忘记移除 `select {}` 使程序挂起      |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "死锁的根本原因（Go 运行时视角）\n",
    "\n",
    "1. **所有存活协程都处于阻塞态**（不含 `time.Sleep`；睡眠结束后仍会继续运行）时，运行时无任务可调度，触发 `fatal error: all goroutines are asleep - deadlock!`。\n",
    "2. **阻塞点缺乏“解除者”**：例如写入满 channel、等待 `WaitGroup`、二次加锁等，需要另一协程协助解除；若协程未启动、提前退出或顺序错误，都可能导致死锁。\n",
    "3. **协程存活≠协程有效**：即使某协程短暂活跃，但在关键时刻如果它已经结束，阻塞点仍无法解开。\n",
    "4. **time.Sleep 不是死锁源**：因为它会在时间到后自动恢复运行；Go 的死锁检测忽略这种自动解除型阻塞。\n",
    "5. **Channel 遍历与关闭协议**：消费者若遍历未关闭的 channel，且无其他写入协程，最终会因等待数据而死锁。协议约定“谁来关闭”极其重要。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "死锁根本原因解析与防护策略\n",
    "\n",
    "1. **“阻塞点 + 无解除者”** 是死锁本质。任何需要外部事件解除的阻塞，都必须确保解除者一定会执行。\n",
    "2. **协程启动顺序至关重要**：例如先调用 `wg.Wait()` 再启动 `Done()` 协程，主协程会在 `Wait()` 处阻塞，无法继续启动新协程。\n",
    "3. **Channel 必须协议清晰**：谁写入、谁关闭、谁消费都要有设计约定。遍历未关闭 channel 会永久阻塞。\n",
    "4. **检查潜在 `panic`**：写入已关闭 channel、重复关闭 channel 等会直接 `panic`，导致协程退出，从而缺少“解除者”。\n",
    "5. **监控与调试**：使用 `go run -race`、`pprof`、日志等工具尽早发现阻塞/死锁；将阻塞点封装成可监控的组件。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### time.Sleep 自动解除阻塞（不会产生死锁）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `time.Sleep` 是“自恢复”阻塞，不依赖其他协程，不会触发死锁。\n",
    "- 常用于节流、等待条件或示意性的延迟；不可用于等待外部事件（会浪费时间）。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "开始 1 秒阻塞\n",
      "sleep over\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tfmt.Println(\"开始 1 秒阻塞\")\n",
    "\ttime.Sleep(1 * time.Second) // 自动解除的阻塞方式：时间到即退出阻塞\n",
    "\tfmt.Println(\"sleep over\")  // 永远能执行到这里，不存在死锁风险\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### sync.WaitGroup 未配套 Done 导致永久阻塞"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `Wait()` 依赖其他协程调用 `Done()`；否则会永久等待直到触发死锁。\n",
    "- 建议搭配 `defer wg.Done()`，确保无论何种流程都能正确归还计数。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "等待子协程结束...\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "fatal error: all goroutines are asleep - deadlock!\n",
      "\n",
      "goroutine 1 [sync.WaitGroup.Wait]:\n",
      "sync.runtime_SemacquireWaitGroup(0x19?)\n",
      "\t/usr/local/go/src/runtime/sema.go:110 +0x25\n",
      "sync.(*WaitGroup).Wait(0x4db2a8?)\n",
      "\t/usr/local/go/src/sync/waitgroup.go:118 +0x48\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [7] Line 21 ]]\u001b[0m /tmp/gonb_2ce7dd61/main.go:20 +0x7b\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "// 运行时可能在所有协程阻塞后报 fatal error: all goroutines are asleep - deadlock!\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar wg sync.WaitGroup\n",
    "\n",
    "\twg.Add(1) // 计数器 +1，表示有一个子任务需要完成\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(500 * time.Millisecond)\n",
    "\t\t// 故意不调用 wg.Done()，导致计数器无法归零\n",
    "\t}()\n",
    "\n",
    "\tfmt.Println(\"等待子协程结束...\")\n",
    "\twg.Wait() // 阻塞等待计数器归零，但由于缺少 Done，这里永久阻塞\n",
    "\tfmt.Println(\"不会打印\") // 永远不会执行到这里\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 从空 Channel 读取导致死锁"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 从空 channel 读取需要“写端协程”配合；若写入发生在阻塞之后才启动，也会死锁。\n",
    "- 正确做法：确保在读取之前开启写协程，或使用缓冲 channel 并写入数据。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "尝试从空 channel 读取...\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "fatal error: all goroutines are asleep - deadlock!\n",
      "\n",
      "goroutine 1 [chan receive]:\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [8] Line 8 ]]\u001b[0m /tmp/gonb_2ce7dd61/main.go:11 +0x7d\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "// 运行结果：阻塞数秒后报 fatal error: all goroutines are asleep - deadlock!\n",
    "\n",
    "package main\n",
    "\n",
    "import \"fmt\"\n",
    "\n",
    "func main() {\n",
    "\tch := make(chan int) // 无缓冲 channel，读写必须配对\n",
    "\tfmt.Println(\"尝试从空 channel 读取...\")\n",
    "\tval := <-ch // 没有任何协程向 ch 写入，会永久阻塞\n",
    "\tfmt.Println(\"永远不会打印:\", val)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### sync.Mutex 二次加锁引发自锁死锁"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `Mutex` 不可重入；同一协程必须先 `Unlock` 才能再次 `Lock`。\n",
    "- 若确实需要可重入行为，可结合 `sync.RWMutex` + 线程 ID 管理，或设计更高层的锁机制。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "第一次获取锁成功\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "fatal error: all goroutines are asleep - deadlock!\n",
      "\n",
      "goroutine 1 [sync.Mutex.Lock]:\n",
      "internal/sync.runtime_SemacquireMutex(0xc000068028?, 0x10?, 0x19?)\n",
      "\t/usr/local/go/src/runtime/sema.go:95 +0x25\n",
      "internal/sync.(*Mutex).lockSlow(0xc0000100f0)\n",
      "\t/usr/local/go/src/internal/sync/mutex.go:149 +0x15d\n",
      "internal/sync.(*Mutex).Lock(...)\n",
      "\t/usr/local/go/src/internal/sync/mutex.go:70\n",
      "sync.(*Mutex).Lock(...)\n",
      "\t/usr/local/go/src/sync/mutex.go:46\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [9] Line 15 ]]\u001b[0m /tmp/gonb_2ce7dd61/main.go:16 +0xa5\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "// 运行结果：阻塞并最终触发死锁错误。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar mu sync.Mutex\n",
    "\n",
    "\tmu.Lock()\n",
    "\tfmt.Println(\"第一次获取锁成功\")\n",
    "\n",
    "\t// 同一协程再次尝试获取同一把互斥锁，会阻塞等待自己释放\n",
    "\tmu.Lock()\n",
    "\tfmt.Println(\"永远不会打印：第二次获取锁成功\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 空 select 永久阻塞"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 空 `select` 阻塞不可恢复，只能通过外部中断结束程序。\n",
    "- 调试时如需暂时阻塞，可使用 `select { case <-done: }` 等可控方式。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "程序将阻塞在空 select 上\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "fatal error: all goroutines are asleep - deadlock!\n",
      "\n",
      "goroutine 1 [select (no cases)]:\n",
      "main.main()\n",
      "\t \u001b[7m[[ Cell [10] Line 7 ]]\u001b[0m /tmp/gonb_2ce7dd61/main.go:10 +0x4b\n",
      "exit status 2\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import \"fmt\"\n",
    "\n",
    "func main() {\n",
    "\tfmt.Println(\"程序将阻塞在空 select 上\")\n",
    "\tselect {} // 没有任何 case 和 default，永远阻塞\n",
    "\t// 空 select 常用作“保持协程存活”的简易方式，但若忘记退出则导致挂起\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 97 用channel实现广播和CountDownLatch"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 场景                         | 核心思路                                                  | Channel 形态                               | 协程行为                                                     | 易踩坑点                                                     |\n",
    "| ---------------------------- | --------------------------------------------------------- | ------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |\n",
    "| 广播信号（Broadcast）        | 关闭同一条 Channel，所有监听方同时获知“出发”事件          | 无缓冲或缓冲均可，通常选 `chan struct{}`   | 发送方只需 `close(ch)`；接收方阻塞在 `<-ch` 或 `range ch`，通道关闭后立即解除阻塞 | 循环变量要以参数形式传入协程，避免闭包捕获同一变量；不要在通道关闭后再写入 |\n",
    "| CountDownLatch（倒计时门闩） | 用有缓冲 Channel 累积“完成信号”，主协程消费固定次数后继续 | 容量设为任务数 P：`make(chan struct{}, P)` | 子协程执行完任务后写入 `ch <- struct{}{}`；主协程循环 `P` 次读取 | 必须确保每个子协程都发送一次信号；容量不足会导致写入阻塞；与 WaitGroup 一样要注意协程启动顺序 |\n",
    "| 空结构体元素                 | 只需靠阻塞/关闭语义，不传输数据                           | `struct{}{}` 零大小、零 GC 负担            | 信号传递而非数据传递                                         | 语法写法为 `struct{}{}`（两对花括号）                        |\n",
    "| 与 `sync.WaitGroup` 对比     | Channel 版可视为“原语级”实现                              | Channel 是更通用的同步工具                 | WaitGroup 封装更简洁但表达力没 Channel 高                    | 选择取决于团队习惯与代码可读性                               |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 通道关闭实现广播通知"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `close(ch)` 是广播信号，所有被 `<-ch` 阻塞的协程同时返回零值（`struct{}{}`）。\n",
    "- 接收方无需消费具体数据，只需感知通道已关闭。\n",
    "- 一旦关闭，禁止再写入，否则会 `panic: send on closed channel`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "乘客 2 等待出发信号...\n",
      "乘客 0 等待出发信号...\n",
      "乘客 1 等待出发信号...\n",
      "领队：所有人准备好了，广播出发信号！\n",
      "乘客 2 收到信号，开始出发！\n",
      "乘客 0 收到信号，开始出发！\n",
      "乘客 1 收到信号，开始出发！\n",
      "所有乘客都已出发。\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// broadcast 通知三名乘客统一出发。\n",
    "// 关键点：关闭 channel 会让所有正在等待的接收方同时解除阻塞。\n",
    "func broadcast() {\n",
    "\tconst passengers = 3\n",
    "\n",
    "\tch := make(chan struct{}) // 无缓冲即可，关注的是关闭语义\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(passengers)\n",
    "\n",
    "\tfor i := 0; i < passengers; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tfmt.Printf(\"乘客 %d 等待出发信号...\\n\", id)\n",
    "\t\t\t<-ch // 阻塞等待广播；通道被关闭时返回零值但不阻塞\n",
    "\t\t\tfmt.Printf(\"乘客 %d 收到信号，开始出发！\\n\", id)\n",
    "\t\t}(i) // 将循环变量作为参数传入，避免闭包捕获问题\n",
    "\t}\n",
    "\n",
    "\ttime.Sleep(2 * time.Second)\n",
    "\tfmt.Println(\"领队：所有人准备好了，广播出发信号！\")\n",
    "\tclose(ch) // 关闭即广播：所有等待协程同时解除阻塞\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"所有乘客都已出发。\")\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tbroadcast()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Channel 模拟 CountDownLatch"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 通道容量为任务数量，确保每次写入不会阻塞。\n",
    "- 主协程读取固定次数即可继续；逻辑上等同于 Java 的 CountDownLatch。\n",
    "- 若某个子协程漏写信号（或 panic），主协程会永久阻塞，需确保任务完成逻辑可靠。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "工人 0 完成任务\n",
      "工人 1 完成任务\n",
      "工人 2 完成任务\n",
      "全部工人已完成，主协程开始执行后续流程。\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// countDownLatch 等待多个工人完成工作后再继续。\n",
    "// 设计：channel 容量等于工人数量，主协程读取 P 次信号再继续执行。\n",
    "func countDownLatch() {\n",
    "\tconst workers = 3\n",
    "\t// 容量等于 workers，避免子协程写入时阻塞（除非写入次数超出）\n",
    "\tdoneCh := make(chan struct{}, workers)\n",
    "\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\t// 模拟不同耗时的任务\n",
    "\t\t\ttime.Sleep(time.Duration(id) * time.Second)\n",
    "\t\t\tfmt.Printf(\"工人 %d 完成任务\\n\", id)\n",
    "\t\t\tdoneCh <- struct{}{} // 将完成信号写入通道\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\t// 主协程阻塞读取 workers 次，确保所有子协程都已完成\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\t<-doneCh\n",
    "\t}\n",
    "\tfmt.Println(\"全部工人已完成，主协程开始执行后续流程。\")\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tcountDownLatch()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 98 招人嫌的sync.Cond"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 主题                   | 核心要点                                                     | 必备前提                                              | 常见坑点                                                     | 实战建议                                                    |\n",
    "| ---------------------- | ------------------------------------------------------------ | ----------------------------------------------------- | ------------------------------------------------------------ | ----------------------------------------------------------- |\n",
    "| 基本概念               | `sync.Cond` 是 Go 中的条件变量，允许 goroutine 在共享状态未满足时等待，并由其它 goroutine 发送信号唤醒 | 需要与互斥锁（`sync.Mutex` / `sync.RWMutex`）配合使用 | 忘记在 `Wait` 前后获取/释放锁；信号在接收方未等待时发送导致“丢信号” | 将共享状态检查放在 `for` 循环里，等待前后的锁粒度一致       |\n",
    "| 阻塞与唤醒流程         | 典型模式：锁定 → 条件不满足则 `Wait` → 被唤醒后重新检查条件 → 满足时继续执行 | `Wait` 内部会暂时解锁，唤醒后重新上锁返回             | 未在 `Wait` 前持有锁会 panic；唤醒后不重新检查条件导致逻辑错误 | 总是搭配 `for !condition { cond.Wait() }`                   |\n",
    "| 单播信号 (`Signal`)    | `Signal()` 唤醒**一个**等待中的 goroutine                    | 有等待者存在                                          | 如果调用时没有 goroutine 在等待，信号丢失                    | 在发送信号前确认等待者已进入 `Wait`                         |\n",
    "| 广播信号 (`Broadcast`) | `Broadcast()` 唤醒**所有**等待中的 goroutine                 | 有等待者存在                                          | 若等待者尚未入场，广播同样会丢失                             | 适合需要唤醒所有工作线程的场景                              |\n",
    "| 与 Channel 对比        | 条件变量可多次广播，适合复杂状态同步；Channel 更简洁，适合一次性通知 | —                                                     | Channel 关闭只能广播一次；条件变量需要更严密的锁管理         | 简单通知优先用 Channel，复杂状态/重复广播再考虑 `sync.Cond` |\n",
    "| 死锁风险               | 所有 goroutine 都在等待信号且没有发送者时死锁                | `Signal`/`Broadcast` 调用者退出或未执行               | 锁竞争、变量状态判断不严谨                                   | 为发送端建立“兜底”逻辑：例如超时或退出时主动广播            |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `sync.Cond` vs Channel：何时选用？\n",
    "\n",
    "| 需求                       | 推荐方案                        | 理由                                                |\n",
    "| -------------------------- | ------------------------------- | --------------------------------------------------- |\n",
    "| 简单的一次性通知           | Channel（单向或带缓冲）         | 写入一次即通知接收方，无需锁管理                    |\n",
    "| 需要传递实际数据           | Channel                         | Channel 具备 FIFO 队列语义，可直接发送数据          |\n",
    "| 需要多次广播或复杂条件判断 | `sync.Cond`                     | 可以重复 `Broadcast` / `Signal`，等待条件可复杂组合 |\n",
    "| 等待某个状态满足才能继续   | `sync.Cond` 或 Channel + select | 若状态受多 goroutine 修改，用 `sync.Cond` 更贴合    |\n",
    "| 控制并发访问、互斥与条件   | `sync.Cond`                     | 结合互斥锁精确控制临界区                            |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 使用 sync.Cond 实现生产者-消费者（单播信号）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `Wait` 必须写在 `for` 循环中，防止“虚假唤醒”或被其它协程抢先消费。\n",
    "- 发送信号前要确保消费者已经等待，否则单播信号会丢失。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[consumer] buffer empty, wait for data\n",
      "[producer] produced item 42\n",
      "[consumer] got item: 42\n",
      "all done\n"
     ]
    }
   ],
   "source": [
    "// 功能：一个消费者等待条件满足（缓冲区有数据）才开始工作；生产者写入后发送信号唤醒消费者。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tbuffer := make([]int, 0, 1)   // 简化：缓冲区只允许一个元素\n",
    "\tmu := sync.Mutex{}            // 保护 buffer\n",
    "\tcond := sync.NewCond(&mu)     // 条件变量与 mutex 绑定\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(2)\n",
    "\n",
    "\t// 消费者：等待缓冲区有数据再消费\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\n",
    "\t\tmu.Lock()\n",
    "\t\tfor len(buffer) == 0 { // 条件不满足则阻塞等待\n",
    "\t\t\tfmt.Println(\"[consumer] buffer empty, wait for data\")\n",
    "\t\t\tcond.Wait() // Wait 会在内部 Unlock，并在被唤醒后重新 Lock\n",
    "\t\t}\n",
    "\t\titem := buffer[0]\n",
    "\t\tbuffer = buffer[:0] // 清空缓冲区\n",
    "\t\tfmt.Println(\"[consumer] got item:\", item)\n",
    "\t\tmu.Unlock()\n",
    "\t}()\n",
    "\n",
    "\t// 生产者：填充数据并发送信号\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\n",
    "\t\ttime.Sleep(500 * time.Millisecond) // 保证消费者先进入等待状态\n",
    "\n",
    "\t\tmu.Lock()\n",
    "\t\tbuffer = append(buffer, 42)\n",
    "\t\tfmt.Println(\"[producer] produced item 42\")\n",
    "\t\tcond.Signal() // 唤醒一个等待中的 goroutine（消费者）\n",
    "\t\tmu.Unlock()\n",
    "\t}()\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"all done\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### sync.Cond 广播实现“一致起跑”"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- **互斥锁 + 条件变量**：`sync.Mutex` 与 `sync.Cond` 保证对共享状态 `currentRound` 的安全访问，并让参赛者在条件不满足时阻塞等待。\n",
    "- **轮次控制**：`currentRound` 记录裁判已广播的最大轮次，参赛者只有在 `currentRound >= 自己应参加的轮次` 时才会继续执行。\n",
    "- **并发同步**：`sync.WaitGroup` 确保所有 goroutine 完成后主函数再退出。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[referee] 第 1 次广播：各位起跑！\n",
      "[runner 3] 收到第 1 轮起跑信号，冲！\n",
      "[runner 1] 收到第 1 轮起跑信号，冲！\n",
      "[runner 2] 收到第 1 轮起跑信号，冲！\n",
      "[referee] 第 2 次广播：各位起跑！\n",
      "[runner 1] 收到第 2 轮起跑信号，冲！\n",
      "[runner 3] 收到第 2 轮起跑信号，冲！\n",
      "[runner 2] 收到第 2 轮起跑信号，冲！\n",
      "所有轮次完成\n"
     ]
    }
   ],
   "source": [
    "// 条件变量唤醒后，goroutine 会重新尝试获取锁，并再次检查 currentRound。只有条件满足才继续执行，否则回到 Wait()。这种模式可以避免虚假唤醒或重复广播带来的逻辑错误\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tvar (\n",
    "\t\tmu        sync.Mutex      // 保护共享状态 currentRound 的互斥锁\n",
    "\t\tstartCond = sync.NewCond(&mu) // 条件变量：参赛者在此等待裁判广播\n",
    "\t)\n",
    "\n",
    "\tconst (\n",
    "\t\trunners     = 3 // 参赛者数量\n",
    "\t\ttotalRounds = 2 // 总共要跑的轮次\n",
    "\t)\n",
    "\n",
    "\tvar (\n",
    "\t\twg           sync.WaitGroup // 用于等待所有 goroutine 完成\n",
    "\t\tcurrentRound int            // 当前已经广播到的轮次（从 0 开始，第一次广播设为 1）\n",
    "\t)\n",
    "\n",
    "\t// WaitGroup 需要等待所有 runner 和裁判 goroutine\n",
    "\twg.Add(runners + 1)\n",
    "\n",
    "\t// 启动多个参赛者 goroutine\n",
    "\tfor runnerID := 1; runnerID <= runners; runnerID++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done() // goroutine 结束时通知 WaitGroup\n",
    "\n",
    "\t\t\t// 每个参赛者需要完成 totalRounds 轮比赛\n",
    "\t\t\tfor round := 1; round <= totalRounds; round++ {\n",
    "\t\t\t\tmu.Lock()\n",
    "\t\t\t\t// 条件变量的典型用法：循环检查条件是否满足\n",
    "\t\t\t\t// 如果当前轮次还没轮到自己，继续等待广播\n",
    "\t\t\t\tfor currentRound < round {\n",
    "\t\t\t\t\tstartCond.Wait() // 阻塞，释放锁，等待 Broadcast 唤醒\n",
    "\t\t\t\t}\n",
    "\t\t\t\t// 条件满足：裁判已经广播到这一轮\n",
    "\t\t\t\tfmt.Printf(\"[runner %d] 收到第 %d 轮起跑信号，冲！\\n\", id, round)\n",
    "\t\t\t\tmu.Unlock() // 离开临界区，让其他 goroutine 继续\n",
    "\n",
    "\t\t\t\t// 这里可以模拟跑步耗时（非必须）\n",
    "\t\t\t\ttime.Sleep(200 * time.Millisecond)\n",
    "\t\t\t}\n",
    "\t\t}(runnerID)\n",
    "\t}\n",
    "\n",
    "\t// 裁判 goroutine：负责逐轮广播起跑信号\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\n",
    "\t\tfor round := 1; round <= totalRounds; round++ {\n",
    "\t\t\t// 每轮广播前先等待一段时间，模拟真实延迟\n",
    "\t\t\ttime.Sleep(time.Duration(round) * 500 * time.Millisecond)\n",
    "\n",
    "\t\t\tmu.Lock()\n",
    "\t\t\t// 更新当前轮次，表示裁判已经发出了 round 轮的广播\n",
    "\t\t\tcurrentRound = round\n",
    "\t\t\tfmt.Printf(\"[referee] 第 %d 次广播：各位起跑！\\n\", round)\n",
    "\t\t\t// 唤醒所有在条件变量上等待的参赛者\n",
    "\t\t\tstartCond.Broadcast()\n",
    "\t\t\tmu.Unlock()\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 等待所有参赛者和裁判完成\n",
    "\twg.Wait()\n",
    "\tfmt.Println(\"所有轮次完成\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 99 MPG并发模型"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [Golang修养之路: 2.Golang的协程调度器原理及GMP设计思想](https://github.com/aceld/golang/blob/main/2%E3%80%81Golang%E7%9A%84%E5%8D%8F%E7%A8%8B%E8%B0%83%E5%BA%A6%E5%99%A8%E5%8E%9F%E7%90%86%E5%8F%8AGMP%E8%AE%BE%E8%AE%A1%E6%80%9D%E6%83%B3%EF%BC%9F.md), [本地](./99/2、Golang的协程调度器原理及GMP设计思想？.pdf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 元素               | 英文含义           | 扮演角色 | 核心职责                                                  | 典型数量级                                    |\n",
    "| ------------------ | ------------------ | -------- | --------------------------------------------------------- | --------------------------------------------- |\n",
    "| **G（Goroutine）** | 轻量级协程描述符   | 任务单位 | 持有栈、指令指针、状态（就绪 / 运行 / 阻塞）、所属 P      | 数量可达百万级                                |\n",
    "| **P（Processor）** | 调度器的本地上下文 | 调度权   | 管理本地运行队列、缓存、协程创建；绑定一个 M 才能执行     | 默认值等于 CPU 核数，可通过 `GOMAXPROCS` 调整 |\n",
    "| **M（Machine）**   | OS 线程            | 执行实体 | 真正在 OS 层面运行 goroutine 的载体；负责系统调用、cgo 等 | 动态创建 / 回收，数量介于 P 与 goroutine 之间 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "调度器工作流程（简要）\n",
    "\n",
    "1. **就绪队列**：每个 P 持有一个 G 本地队列（LQ），另有全局队列（GQ）。调度优先从本地取 G，若空则偷取其他 P 的一半任务（Work-Stealing）。\n",
    "2. **创建与唤醒**：`go func()` 新建的 G 优先入当前 P 的 LQ；被唤醒的 G 若无可用 P，会放入 GQ。\n",
    "3. **阻塞处理**：\n",
    "   - **协作式阻塞**：如 `time.Sleep`、`channel`、`mutex` 等，调度器会将当前 G 置为等待态并立即调度下一个 G。\n",
    "   - **系统调用阻塞**：若 G 在 M 上调用阻塞型 syscalls，P 会暂时解绑并移交给新的 M，以避免整体停摆。\n",
    "4. **抢占机制**：Go1.14 起支持异步抢占。长时间运行的 G 会被调度器中断，以保证公平性。\n",
    "5. **调试入口**：`GODEBUG=schedtrace=1000,scheddetail=1` 可输出调度轨迹；`runtime/metrics` 与 `pprof` 可观察线程、队列等指标。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### GOMAXPROCS 控制 P 数量的影响"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `GOMAXPROCS` 控制可并行运行的 P 数量，从而决定最多有多少个 M 可以被调度执行 goroutine。\n",
    "- 当 P=1 时，即使有多个 G，也只能在单线程上交替运行；P=4 时，可充分利用四核并行。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "逻辑 CPU 数量: 16\n",
      "\n",
      "=== 场景：GOMAXPROCS=1，worker=4 ===\n",
      "[worker-3] done, checksum=208333320833334.88\n",
      "[worker-0] done, checksum=208333320833334.88\n",
      "[worker-1] done, checksum=208333320833334.88\n",
      "[worker-2] done, checksum=208333320833334.88\n",
      "总耗时: 102.140903ms\n",
      "\n",
      "=== 场景：GOMAXPROCS=4，worker=4 ===\n",
      "[worker-3] done, checksum=208333320833334.88\n",
      "[worker-1] done, checksum=208333320833334.88\n",
      "[worker-2] done, checksum=208333320833334.88\n",
      "[worker-0] done, checksum=208333320833334.88\n",
      "总耗时: 27.486421ms\n",
      "\n",
      "并行收益比: 3.72x\n",
      "提示：配合 GODEBUG=schedtrace=1000 运行可观察调度事件。\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// heavyCPUWork 模拟 CPU 密集型任务：计算大量平方根。\n",
    "func heavyCPUWork(id, iterations int, wg *sync.WaitGroup) {\n",
    "\tdefer wg.Done()\n",
    "\n",
    "\tvar sum float64\n",
    "\tfor i := 0; i < iterations; i++ {\n",
    "\t\tsum += float64(i*i) / float64(iterations)\n",
    "\t\tif i%5_000_000 == 0 {\n",
    "\t\t\t// 使用 Gosched() 让出调度机会，便于观察协作式调度。\n",
    "\t\t\truntime.Gosched()\n",
    "\t\t}\n",
    "\t}\n",
    "\tfmt.Printf(\"[worker-%d] done, checksum=%.2f\\n\", id, sum)\n",
    "}\n",
    "\n",
    "// runScenario 在指定 P（GOMAXPROCS）数量下运行 N 个 goroutine。\n",
    "func runScenario(procs, workers int) time.Duration {\n",
    "\t// 保存旧值，执行完恢复，避免影响后续实验。\n",
    "\tprev := runtime.GOMAXPROCS(procs)\n",
    "\tdefer runtime.GOMAXPROCS(prev)\n",
    "\n",
    "\tfmt.Printf(\"\\n=== 场景：GOMAXPROCS=%d，worker=%d ===\\n\", procs, workers)\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(workers)\n",
    "\n",
    "\tstart := time.Now()\n",
    "\tfor i := 0; i < workers; i++ {\n",
    "\t\tgo heavyCPUWork(i, 25_000_000, &wg)\n",
    "\t}\n",
    "\twg.Wait()\n",
    "\telapsed := time.Since(start)\n",
    "\n",
    "\tfmt.Printf(\"总耗时: %v\\n\", elapsed)\n",
    "\treturn elapsed\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tfmt.Printf(\"逻辑 CPU 数量: %d\\n\", runtime.NumCPU())\n",
    "\tone := runScenario(1, 4) // 只有一个 P，所有 CPU 任务必须串行执行\n",
    "\tmulti := runScenario(4, 4) // P=4 时可四核并行，耗时大幅下降\n",
    "\n",
    "\tfmt.Printf(\"\\n并行收益比: %.2fx\\n\", float64(one)/float64(multi))\n",
    "\tfmt.Println(\"提示：配合 GODEBUG=schedtrace=1000 运行可观察调度事件。\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### MPG 如何处理阻塞以及网络 I/O"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 虽然 `GOMAXPROCS=1`，但阻塞在网络 I/O 的 G 会释放 P 与 M 的绑定；调度器复用 / 新建 OS 线程继续执行其他 G。\n",
    "- `slowHandler` 中的 `time.Sleep` 并不会阻塞整个线程，因为 runtime 将该 goroutine 停放，P 会继续为可运行的 goroutine 提供服务。\n",
    "- 可搭配 `GODEBUG=scheddetail=1,schedtrace=1000 go run main.go` 观察更多细节（线程创建、P 绑定情况）。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "GOMAXPROCS 被设置为 1，仅保留单个 P\n",
      "[net] 开始发起慢速 HTTP 请求\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[net] 请求结束，响应: slow response\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] still alive (G 继续在 P 上运行)\n",
      "[spinner] stop signal received\n",
      "观察：即便网络 goroutine 阻塞在系统调用，spinner 依旧持续运行\n",
      "原因：阻塞的 G 被移出 P，runtime 另派 M 协助执行可运行的 G，保持调度前进。\n"
     ]
    }
   ],
   "source": [
    "// 在只有一个 P 的情况下，让某个 goroutine 发起网络请求（会触发阻塞系统调用），观察 CPU goroutine 仍能执行，从侧面验证“阻塞时 G 会脱离 P，由新的 M 接管”的调度行为。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"io\"\n",
    "\t\"net/http\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// slowHandler 模拟慢速服务端：处理请求时休眠 2 秒。\n",
    "func slowHandler(w http.ResponseWriter, r *http.Request) {\n",
    "\ttime.Sleep(2 * time.Second)\n",
    "\tio.WriteString(w, \"slow response\")\n",
    "}\n",
    "\n",
    "func startServer(addr string, ready chan<- struct{}) {\n",
    "\thttp.HandleFunc(\"/slow\", slowHandler)\n",
    "\n",
    "\tgo func() {\n",
    "\t\tclose(ready)                          // 通知主程序服务器已启动\n",
    "\t\tif err := http.ListenAndServe(addr, nil); err != nil {\n",
    "\t\t\tpanic(err)\n",
    "\t\t}\n",
    "\t}()\n",
    "}\n",
    "\n",
    "// cpuSpinner 连续打印心跳，模拟 CPU 密集型 goroutine。\n",
    "func cpuSpinner(stop <-chan struct{}, wg *sync.WaitGroup) {\n",
    "\tdefer wg.Done()\n",
    "\n",
    "\tticker := time.NewTicker(200 * time.Millisecond)\n",
    "\tdefer ticker.Stop()\n",
    "\n",
    "\tfor {\n",
    "\t\tselect {\n",
    "\t\tcase <-ticker.C:\n",
    "\t\t\tfmt.Println(\"[spinner] still alive (G 继续在 P 上运行)\")\n",
    "\t\tcase <-stop:\n",
    "\t\t\tfmt.Println(\"[spinner] stop signal received\")\n",
    "\t\t\treturn\n",
    "\t\t}\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\taddr := \"127.0.0.1:8080\"\n",
    "\tready := make(chan struct{})\n",
    "\tstartServer(addr, ready)\n",
    "\t<-ready // 等待 HTTP 服务器启动完成\n",
    "\n",
    "\tprev := runtime.GOMAXPROCS(1) // 限制只有 1 个 P，让调度更容易观察\n",
    "\tdefer runtime.GOMAXPROCS(prev)\n",
    "\n",
    "\tfmt.Println(\"GOMAXPROCS 被设置为 1，仅保留单个 P\")\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\tstop := make(chan struct{})\n",
    "\twg.Add(1)\n",
    "\tgo cpuSpinner(stop, &wg) // 启动 CPU goroutine\n",
    "\n",
    "\ttime.Sleep(100 * time.Millisecond)\n",
    "\n",
    "\t// 另开 goroutine 执行阻塞网络请求。\n",
    "\twg.Add(1)\n",
    "\tgo func() {\n",
    "\t\tdefer wg.Done()\n",
    "\t\tfmt.Println(\"[net] 开始发起慢速 HTTP 请求\")\n",
    "\t\tresp, err := http.Get(\"http://\" + addr + \"/slow\")\n",
    "\t\tif err != nil {\n",
    "\t\t\tfmt.Println(\"[net] 请求失败:\", err)\n",
    "\t\t\treturn\n",
    "\t\t}\n",
    "\t\tdefer resp.Body.Close()\n",
    "\t\tbody, _ := io.ReadAll(resp.Body)\n",
    "\t\tfmt.Println(\"[net] 请求结束，响应:\", string(body))\n",
    "\t}()\n",
    "\n",
    "\ttime.Sleep(3 * time.Second) // 等待网络请求完成\n",
    "\tclose(stop)\n",
    "\twg.Wait()\n",
    "\n",
    "\tfmt.Println(\"观察：即便网络 goroutine 阻塞在系统调用，spinner 依旧持续运行\")\n",
    "\tfmt.Println(\"原因：阻塞的 G 被移出 P，runtime 另派 M 协助执行可运行的 G，保持调度前进。\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 100 协程与线程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 维度        | Goroutine（协程）                     | OS Thread（线程）                   | 备注                                 |\n",
    "| ----------- | ------------------------------------- | ----------------------------------- | ------------------------------------ |\n",
    "| 创建/调度者 | Go 运行时调度器（MPG 模型）           | 操作系统内核                        | Go 自带用户态调度，减少内核交互      |\n",
    "| 栈空间      | 初始≈2 KB，按需自动增长/收缩          | 默认 1 MB（可调），固定分配         | 协程更易大量创建                     |\n",
    "| 上下文切换  | 用户态完成，大多不陷入内核            | 内核态调度，切换成本高              | Goroutine 切换更轻量                 |\n",
    "| 并发数量    | 可轻松达数十万甚至百万                | 受内核/内存限制，通常千级以内       | Go 通过复用线程实现海量并发          |\n",
    "| 阻塞行为    | 调度器感知阻塞并调度其他 G            | 阻塞会占用线程资源                  | Go runtime 会将阻塞 G 移出 P 队列    |\n",
    "| 同步原语    | Channel、WaitGroup、Mutex 等          | Mutex、Condition、Semaphore 等      | Go 同步原语构建在 goroutine 调度之上 |\n",
    "| 线程亲和性  | 默认不保证，与 M 动态绑定             | 固定线程 ID，具备线程局部存储       | 可用 `runtime.LockOSThread` 绑定     |\n",
    "| 启动方式    | `go func() { … }()`                   | `pthread_create` / `std::thread` 等 | Goroutine 语法极简                   |\n",
    "| 调试指标    | `runtime.NumGoroutine`、`scheddetail` | `top`/`ps`/`perf` 等工具            | 需结合 runtime 工具观察              |\n",
    "\n",
    "**核心 takeaway：** Goroutine 是 Go runtime 管理的用户态“绿色线程”，通过少量 OS 线程复用实现海量并发；OS 线程由内核调度，创建与上下文切换成本高，数量受限。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 协程的轻量级并发能力"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Goroutine 创建、调度几乎瞬间完成；即使数量达十万也不会明显拖慢程序。\n",
    "- `runtime.Gosched()` 用于演示协作式调度，让其它 goroutine 获得执行机会。\n",
    "- `atomic.AddUint64` 确保并发安全地累加计数结果。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "逻辑 CPU 数量: 16\n",
      "实际执行 goroutine 数量: 100000000\n",
      "成功执行计数: 100000000\n",
      "总耗时: 16.328348383s\n",
      "当前运行中的 goroutine: 1\n"
     ]
    }
   ],
   "source": [
    "// 启动 1亿 个 goroutine，验证创建和调度成本极低，并统计执行结果。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"sync/atomic\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tfmt.Printf(\"逻辑 CPU 数量: %d\\n\", runtime.NumCPU())\n",
    "\n",
    "\tconst goroutines = 100_000_000       // 演示海量 goroutine\n",
    "\tvar counter uint64               // 原子计数器，用于验证执行次数\n",
    "\tvar wg sync.WaitGroup            // 同步所有 goroutine\n",
    "\n",
    "\twg.Add(goroutines)\n",
    "\tstart := time.Now()\n",
    "\n",
    "\tfor i := 0; i < goroutines; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\n",
    "\t\t\t// 模拟一些工作：原子自增，代表任务被执行了一次。\n",
    "\t\t\tatomic.AddUint64(&counter, 1)\n",
    "\n",
    "\t\t\t// 每隔一定数量主动让出调度机会，演示协作式抢占。\n",
    "\t\t\tif id%10_000 == 0 {\n",
    "\t\t\t\truntime.Gosched()\n",
    "\t\t\t}\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\telapsed := time.Since(start)\n",
    "\n",
    "\tfmt.Printf(\"实际执行 goroutine 数量: %d\\n\", goroutines)\n",
    "\tfmt.Printf(\"成功执行计数: %d\\n\", counter)\n",
    "\tfmt.Printf(\"总耗时: %v\\n\", elapsed)\n",
    "\tfmt.Printf(\"当前运行中的 goroutine: %d\\n\", runtime.NumGoroutine())\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "逻辑 CPU 数量: 16\n",
      "实际执行 goroutine 数量: 100000\n",
      "成功执行计数: 100000\n",
      "总耗时: 18.954135ms\n",
      "当前运行中的 goroutine: 1\n"
     ]
    }
   ],
   "source": [
    "// 启动 10 万个 goroutine，验证创建和调度成本极低，并统计执行结果。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"sync/atomic\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tfmt.Printf(\"逻辑 CPU 数量: %d\\n\", runtime.NumCPU())\n",
    "\n",
    "\tconst goroutines = 100_000       // 演示海量 goroutine\n",
    "\tvar counter uint64               // 原子计数器，用于验证执行次数\n",
    "\tvar wg sync.WaitGroup            // 同步所有 goroutine\n",
    "\n",
    "\twg.Add(goroutines)\n",
    "\tstart := time.Now()\n",
    "\n",
    "\tfor i := 0; i < goroutines; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\n",
    "\t\t\t// 模拟一些工作：原子自增，代表任务被执行了一次。\n",
    "\t\t\tatomic.AddUint64(&counter, 1)\n",
    "\n",
    "\t\t\t// 每隔一定数量主动让出调度机会，演示协作式抢占。\n",
    "\t\t\tif id%10_000 == 0 {\n",
    "\t\t\t\truntime.Gosched()\n",
    "\t\t\t}\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\telapsed := time.Since(start)\n",
    "\n",
    "\tfmt.Printf(\"实际执行 goroutine 数量: %d\\n\", goroutines)\n",
    "\tfmt.Printf(\"成功执行计数: %d\\n\", counter)\n",
    "\tfmt.Printf(\"总耗时: %v\\n\", elapsed)\n",
    "\tfmt.Printf(\"当前运行中的 goroutine: %d\\n\", runtime.NumGoroutine())\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 协程与 OS 线程的绑定与差异"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 使用 `runtime.LockOSThread()` 可以确保某个 goroutine 始终在同一 OS 线程上运行（常用于与线程局部状态 / GUI / cgo 交互）。\n",
    "- pthread_self` 返回线程 ID，验证 goroutine 与线程的一对多关系：不锁定时 goroutine 可在不同线程间迁移。\n",
    "- 即使绑定了线程，调度与栈管理仍由 Go runtime 负责；无法直接管理线程栈大小、调度策略等。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n"
     ]
    }
   ],
   "source": [
    "! go env CGO_ENABLED"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[worker-3] 已绑定 OS 线程 ID=134570839357248\n",
      "[worker-3] 第 1 轮工作，仍在 OS 线程 134570839357248 上\n",
      "[worker-1] 已绑定 OS 线程 ID=134569625896512\n",
      "[worker-1] 第 1 轮工作，仍在 OS 线程 134569625896512 上\n",
      "[worker-0] 已绑定 OS 线程 ID=134569642681920\n",
      "[worker-0] 第 1 轮工作，仍在 OS 线程 134569642681920 上\n",
      "[worker-2] 已绑定 OS 线程 ID=134569634289216\n",
      "[worker-2] 第 1 轮工作，仍在 OS 线程 134569634289216 上\n",
      "[worker-3] 第 2 轮工作，仍在 OS 线程 134570839357248 上\n",
      "[worker-0] 第 2 轮工作，仍在 OS 线程 134569642681920 上\n",
      "[worker-2] 第 2 轮工作，仍在 OS 线程 134569634289216 上\n",
      "[worker-1] 第 2 轮工作，仍在 OS 线程 134569625896512 上\n",
      "[worker-1] 第 3 轮工作，仍在 OS 线程 134569625896512 上\n",
      "[worker-2] 第 3 轮工作，仍在 OS 线程 134569634289216 上\n",
      "[worker-0] 第 3 轮工作，仍在 OS 线程 134569642681920 上\n",
      "[worker-3] 第 3 轮工作，仍在 OS 线程 134570839357248 上\n",
      "[worker-3] 完成所有工作，退出时仍绑定 OS 线程 134570839357248\n",
      "[worker-1] 完成所有工作，退出时仍绑定 OS 线程 134569625896512\n",
      "[worker-0] 完成所有工作，退出时仍绑定 OS 线程 134569642681920\n",
      "[worker-2] 完成所有工作，退出时仍绑定 OS 线程 134569634289216\n",
      "所有绑定线程的 goroutine 执行完毕。\n"
     ]
    }
   ],
   "source": [
    "! cd 100 && go run ."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 101 用channel并行处理海量文件"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 阶段         | 角色                 | 关键职责                                                 | 并发控制手段                                     | 背压机制                                   |\n",
    "| ------------ | -------------------- | -------------------------------------------------------- | ------------------------------------------------ | ------------------------------------------ |\n",
    "| **目录遍历** | 生产者 (walkers)     | 递归扫描目录、筛选常规文件                               | 少量 goroutine（通常 1 个） + `filepath.WalkDir` | `fileCh` 满时阻塞，防止一次性塞入过多文件  |\n",
    "| **文件读取** | 中间处理者 (readers) | 打开文件、按行读取、清理行尾                             | 多个 goroutine，从 `fileCh` 中取任务             | `lineCh` 满时阻塞，控制 I/O 与计算速度匹配 |\n",
    "| **行级处理** | 消费者 (processors)  | 解析行内容、执行计算（求和/过滤/聚合）                   | 多个 goroutine，吞吐取决于 CPU                   | `lineCh` 为空时阻塞，等待新数据            |\n",
    "| **监控**     | 辅助 goroutine       | 周期性打印 `len(fileCh)` / `len(lineCh)`                 | 1 个 goroutine + `time.Ticker`                   | 通过观测指标手动调节上下游并发度           |\n",
    "| **同步收尾** | WaitGroup + Channel  | 在正确时机关闭各级 channel，避免“读关闭之后”或“提前关闭” | `sync.WaitGroup`                                 | 对应阶段完成后关闭下一阶段输入管道         |\n",
    "\n",
    "核心思想：将“海量文件 → 行级数据 → 业务处理”拆成三段流水线，用两个带缓冲的 channel 串联，实现高吞吐同时保持资源可控。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 完整可运行的三段式 Channel 流水线"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. **walkDir**\n",
    "   - 递归遍历目录并将常规文件推送到 `fileCh`。\n",
    "   - `fileCh` 缓冲区满时自动阻塞，形成“背压”。\n",
    "   - 异常情况下输出日志不中断流程。\n",
    "2. **readFile**\n",
    "   - 默认开启多协程读取（示例为 12 个，可按磁盘带宽调整）。\n",
    "\n",
    "- 每个 reader 循环消费 `fileCh`，按行读取并写入 `lineCh`。\n",
    "- 在 `fileCh` 被关闭且读取完毕后自动退出。\n",
    "\n",
    "1. **processLine**\n",
    "   - 消费 `lineCh`，将字符串解析为整数；非数字行打印告警。\n",
    "   - 使用 `atomic.Int64`（或 `atomic.AddInt64`）确保并发安全。\n",
    "2. **DealMassFile 设计思路（本示例复刻）**\n",
    "   - 监控 goroutine 每秒打印 `len(fileCh)` / `len(lineCh)`，帮助判断是否需要调整并发度。\n",
    "   - 启动顺序：walk → read → process；关闭顺序严格反向，确保不会“读空 channel”或“向关闭 channel 写入”。\n",
    "   - 根据堆积情况调节：\n",
    "     - **高 backlog + 低 CPU 利用率** → 提升下游（process）并发。\n",
    "     - **高 backlog + CPU 已满** → 降低上游（walk/read）并发。\n",
    "3. **并发容器对比**\n",
    "   - 切片/数组需要额外锁才能安全写。\n",
    "   - map 禁止并发写。\n",
    "   - channel 天然支持安全的多生产者、多消费者模型。\n",
    "4. **channel 设计模式**\n",
    "   - 两级 channel 构成典型的生产线：路径 → 行 → 业务处理。\n",
    "   - 使用缓冲 channel + 监控指标实现水位控制。\n",
    "5. **并发度优化策略**\n",
    "   - 文件数远大于 CPU 核数：限制 goroutine 数，避免上下文切换浪费。\n",
    "   - 文件数量有限且每个文件很大：增加读取协程，减轻单个文件读取延迟。\n",
    "   - 目标是保持 CPU 利用率稳定（例如 80%~90%）。\n",
    "6. **goroutine 分组管理**\n",
    "   - 明确区分三类 goroutine：遍历组、读取组、处理组。\n",
    "   - WaitGroup + channel 关闭顺序确保生命周期安全。\n",
    "7. **性能监控与调优**\n",
    "   - 持续观察 channel backlog、CPU 利用率、吞吐量。\n",
    "   - 根据监控结果决定是扩大下游、减少上游、还是调整缓冲区大小。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "测试数据目录: /tmp/pipeline-demo-3834420253\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "[processor] \"not-a-number\" is not a number\n",
      "总和 = 786886, 耗时 = 1.150369ms\n",
      "提示：若 monitor 输出表明 backlog 持续升高，可适当增加下游并发度。\n"
     ]
    }
   ],
   "source": [
    "// 功能：自动构造测试目录 → 并行遍历/读取/处理 → 统计所有数字行的总和，并演示监控输出\n",
    "// 提示：若在 Windows 或其他平台编译，请确保 fs.ErrClosed 和 io.EOF 等判定逻辑符合本地行为；根据实际 CPU/磁盘性能调整 GOMAXPROCS 与并发数量。\n",
    "\n",
    "package main\n",
    "\n",
    "import (\n",
    "\t\"bufio\"\n",
    "\t\"fmt\"\n",
    "\t\"io/fs\"\n",
    "\t\"math/rand\"\n",
    "\t\"os\"\n",
    "\t\"path/filepath\"\n",
    "\t\"strconv\"\n",
    "\t\"strings\"\n",
    "\t\"sync\"\n",
    "\t\"sync/atomic\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// ---------- 配置参数：根据机器性能调节 ----------\n",
    "\n",
    "const (\n",
    "\tfileWalkerGoroutines  = 1   // 一般 1 个足够；目录极其庞大时可增加\n",
    "\tfileReaderGoroutines  = 12  // I/O 密集，数量可以高一些\n",
    "\tlineWorkerGoroutines  = 6   // CPU 密集，关注 CPU 利用率\n",
    "\tfileChannelBufferSize = 256 // 控制文件任务排队长度\n",
    "\tlineChannelBufferSize = 2048\n",
    "\tmonitorInterval       = time.Second\n",
    ")\n",
    "\n",
    "// pipeline 封装了整个处理流水线运行所需的共享资源。\n",
    "type pipeline struct {\n",
    "\tfileCh chan string\n",
    "\tlineCh chan string\n",
    "\n",
    "\tsum atomic.Int64 //线程安全的累计值\n",
    "\n",
    "\twalkWG    sync.WaitGroup\n",
    "\treadWG    sync.WaitGroup\n",
    "\tprocessWG sync.WaitGroup\n",
    "}\n",
    "\n",
    "// run 负责 orchestrate 全流程：启动各阶段、等待、关闭通道。\n",
    "func (p *pipeline) run(root string) int64 {\n",
    "\t// 监控协程：定期打印队列水位，便于手动调节并发度\n",
    "\tgo func() {\n",
    "\t\tticker := time.NewTicker(monitorInterval)\n",
    "\t\tdefer ticker.Stop()\n",
    "\t\tfor range ticker.C {\n",
    "\t\t\tfmt.Printf(\"[monitor] fileCh backlog=%d, lineCh backlog=%d\\n\",\n",
    "\t\t\t\tlen(p.fileCh), len(p.lineCh))\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// --- 目录遍历阶段 ---\n",
    "\tp.walkWG.Add(fileWalkerGoroutines)\n",
    "\tfor i := 0; i < fileWalkerGoroutines; i++ {\n",
    "\t\tgo p.walkDir(root)\n",
    "\t}\n",
    "\n",
    "\t// --- 文件读取阶段 ---\n",
    "\tp.readWG.Add(fileReaderGoroutines)\n",
    "\tfor i := 0; i < fileReaderGoroutines; i++ {\n",
    "\t\tgo p.readFile()\n",
    "\t}\n",
    "\n",
    "\t// --- 行级处理阶段 ---\n",
    "\tp.processWG.Add(lineWorkerGoroutines)\n",
    "\tfor i := 0; i < lineWorkerGoroutines; i++ {\n",
    "\t\tgo p.processLine()\n",
    "\t}\n",
    "\n",
    "\t// --- 关闭顺序：上游完成 -> 关闭对应 channel -> 下游读完退出 ---\n",
    "\n",
    "\tp.walkWG.Wait()\n",
    "\tclose(p.fileCh) // 通知读取协程没有新文件了\n",
    "\n",
    "\tp.readWG.Wait()\n",
    "\tclose(p.lineCh) // 通知处理协程没有新数据了\n",
    "\n",
    "\tp.processWG.Wait()\n",
    "\treturn p.sum.Load()\n",
    "}\n",
    "\n",
    "// walkDir 递归遍历目录，把常规文件路径写入 fileCh。\n",
    "// 由于 fileCh 是带缓冲 channel，当下游消费慢时会自动阻塞形成背压。\n",
    "func (p *pipeline) walkDir(root string) {\n",
    "\tdefer p.walkWG.Done()\n",
    "\n",
    "\terr := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {\n",
    "\t\tif err != nil {\n",
    "\t\t\t// 遇到异常（例如权限问题），打印后跳过\n",
    "\t\t\tfmt.Printf(\"[walker] skip %s: %v\\n\", path, err)\n",
    "\t\t\treturn nil\n",
    "\t\t}\n",
    "\t\tif d.Type().IsRegular() {\n",
    "\t\t\tp.fileCh <- path\n",
    "\t\t}\n",
    "\t\treturn nil\n",
    "\t})\n",
    "\n",
    "\tif err != nil {\n",
    "\t\tfmt.Printf(\"[walker] walk error: %v\\n\", err)\n",
    "\t}\n",
    "}\n",
    "\n",
    "// readFile 从 fileCh 取文件路径，按行读取到 lineCh。\n",
    "func (p *pipeline) readFile() {\n",
    "\tdefer p.readWG.Done()\n",
    "\n",
    "\tfor path := range p.fileCh {\n",
    "\t\tif err := p.streamFile(path); err != nil {\n",
    "\t\t\tfmt.Printf(\"[reader] %s: %v\\n\", path, err)\n",
    "\t\t}\n",
    "\t}\n",
    "}\n",
    "\n",
    "// streamFile 负责实际打开文件、逐行读取并写入 lineCh。\n",
    "func (p *pipeline) streamFile(path string) error {\n",
    "\tfile, err := os.Open(path)\n",
    "\tif err != nil {\n",
    "\t\treturn err\n",
    "\t}\n",
    "\t// 注意：必须在函数末尾关闭文件，不能放在 goroutine 循环最外层的 defer（否则会延迟关闭）\n",
    "\tdefer file.Close()\n",
    "\n",
    "\treader := bufio.NewReader(file)\n",
    "\tfor {\n",
    "\t\tline, err := reader.ReadString('\\n')\n",
    "\t\tif err != nil {\n",
    "\t\t\tline = strings.TrimSpace(line)\n",
    "\t\t\tif len(line) > 0 {\n",
    "\t\t\t\tp.lineCh <- line\n",
    "\t\t\t}\n",
    "\t\t\tif err == fs.ErrClosed {\n",
    "\t\t\t\treturn err\n",
    "\t\t\t}\n",
    "\t\t\tif err == io.EOF {\n",
    "\t\t\t\treturn nil\n",
    "\t\t\t}\n",
    "\t\t\treturn err\n",
    "\t\t}\n",
    "\t\tp.lineCh <- strings.TrimSpace(line)\n",
    "\t}\n",
    "}\n",
    "\n",
    "// processLine 从 lineCh 取数据，尝试解析为整数并累加。\n",
    "// 使用 atomic 保证并发安全，并为非数字行输出提示。\n",
    "func (p *pipeline) processLine() {\n",
    "\tdefer p.processWG.Done()\n",
    "\n",
    "\tfor line := range p.lineCh {\n",
    "\t\tif len(line) == 0 {\n",
    "\t\t\tcontinue\n",
    "\t\t}\n",
    "\t\tvalue, err := strconv.Atoi(line)\n",
    "\t\tif err != nil {\n",
    "\t\t\tfmt.Printf(\"[processor] %q is not a number\\n\", line)\n",
    "\t\t\tcontinue\n",
    "\t\t}\n",
    "\t\tp.sum.Add(int64(value))\n",
    "\t}\n",
    "}\n",
    "\n",
    "// ---------- 演示用：准备测试数据并运行 ----------\n",
    "\n",
    "func main() {\n",
    "\ttempDir, cleanup := prepareTestData()\n",
    "\tdefer cleanup()\n",
    "\n",
    "\tfmt.Println(\"测试数据目录:\", tempDir)\n",
    "\n",
    "\tp := &pipeline{\n",
    "\t\tfileCh: make(chan string, fileChannelBufferSize),\n",
    "\t\tlineCh: make(chan string, lineChannelBufferSize),\n",
    "\t}\n",
    "\n",
    "\tstart := time.Now()\n",
    "\ttotal := p.run(tempDir)\n",
    "\telapsed := time.Since(start)\n",
    "\n",
    "\tfmt.Printf(\"总和 = %d, 耗时 = %v\\n\", total, elapsed)\n",
    "\tfmt.Println(\"提示：若 monitor 输出表明 backlog 持续升高，可适当增加下游并发度。\")\n",
    "}\n",
    "\n",
    "// prepareTestData 创建一个临时目录，写入若干模拟日志文件，便于自包含示例。\n",
    "// 返回目录路径以及清理函数。\n",
    "func prepareTestData() (string, func()) {\n",
    "\tdir, err := os.MkdirTemp(\"\", \"pipeline-demo-*\")\n",
    "\tif err != nil {\n",
    "\t\tpanic(err)\n",
    "\t}\n",
    "\n",
    "\tr := rand.New(rand.NewSource(time.Now().UnixNano()))\n",
    "\n",
    "\twriteFile := func(name string, lines []string) {\n",
    "\t\tpath := filepath.Join(dir, name)\n",
    "\t\tif err := os.WriteFile(path, []byte(strings.Join(lines, \"\\n\")+\"\\n\"), 0o644); err != nil {\n",
    "\t\t\tpanic(err)\n",
    "\t\t}\n",
    "\t}\n",
    "\n",
    "\t// 生成 20 个小文件，内容包含正整数、空行、非法字符串\n",
    "\tfor i := 0; i < 20; i++ {\n",
    "\t\tvar lines []string\n",
    "\t\tfor j := 0; j < 100; j++ {\n",
    "\t\t\tswitch r.Intn(10) {\n",
    "\t\t\tcase 0:\n",
    "\t\t\t\tlines = append(lines, \"not-a-number\")\n",
    "\t\t\tcase 1:\n",
    "\t\t\t\tlines = append(lines, \"\")\n",
    "\t\t\tdefault:\n",
    "\t\t\t\tlines = append(lines, strconv.Itoa(r.Intn(1000)))\n",
    "\t\t\t}\n",
    "\t\t}\n",
    "\t\twriteFile(fmt.Sprintf(\"file-%02d.log\", i), lines)\n",
    "\t}\n",
    "\n",
    "\treturn dir, func() {\n",
    "\t\tos.RemoveAll(dir)\n",
    "\t}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 102 用channel限制接口的并发请求量"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 关注维度 | 关键要点                                                   | Go 实现技巧                                    | 易忽略风险                                               |\n",
    "| -------- | ---------------------------------------------------------- | ---------------------------------------------- | -------------------------------------------------------- |\n",
    "| 限流动机 | 保护 CPU、内存、外部服务（Redis/MySQL/下游 API），避免雪崩 | 对“瞬时并发”设定硬阈值                         | 限流会引入排队或超时，需评估用户体验                     |\n",
    "| 核心模型 | 使用带缓冲 channel 作为“信号量”/“令牌桶”                   | `sem := make(chan struct{}, N)`                | channel 容量即最大并发；容量过大形同虚设                 |\n",
    "| 请求流程 | 获取令牌 → 执行业务 → 释放令牌                             | `sem <- struct{}{}`、`defer func(){ <-sem }()` | 忘记释放令牌会导致死锁或吞吐崩溃                         |\n",
    "| 阻塞行为 | channel 填满时写操作阻塞，请求自动排队                     | 无需显式锁、条件变量                           | 长时间阻塞需搭配超时/上下文取消                          |\n",
    "| 调优指标 | 队列长度、平均等待时间、CPU/内存/下游负载                  | 可搭配监控打印 `len(sem)`                      | 高等待 + 高负载：扩大资源；高等待 + 低负载：提高并发上限 |\n",
    "| 扩展能力 | 静态阈值简单可靠；结合速率限制或动态阈值更灵活             | 可叠加 `time.Ticker`、滑动窗口等节流手段       | 单独使用 channel 无法平滑速率，仅限制瞬时并发            |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "参考要点归纳\n",
    "\n",
    "1. **限制并发请求量的必要性**\n",
    "   - CPU 保护：避免计算密集任务使 CPU 打满。\n",
    "   - 内存保护：控制大对象创建频率（如临时切片、map）。\n",
    "   - 下游服务保护：防止高并发压垮 Redis/MySQL/第三方 API。\n",
    "   - 代价：请求可能排队或被拒绝，影响响应时间与体验。\n",
    "2. **Channel 并发限流原理**\n",
    "   - 缓冲 channel 作为令牌桶，容量决定瞬时并发上限。\n",
    "   - 请求开始 `sem <- struct{}{}`，结束 `defer func(){ <-sem }()`。\n",
    "   - channel 满时写操作阻塞，自动形成排队。\n",
    "3. **技术细节与最佳实践**\n",
    "   - 令牌类型选 `struct{}`，最省内存。\n",
    "   - 必须使用 `defer` 保证释放，否则会发生死锁/资源枯竭。\n",
    "   - 单纯的 channel 限制只控制瞬时并发，不平滑速率，可与 ticker/滑动窗口组合。\n",
    "   - 对等待过长的请求，应设置超时或降级策略。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 最小实现 —— 并发信号量控制 100 个并发"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `semaphore <- struct{}{}` 会在缓冲区满时阻塞，从而防止超过设定并发。\n",
    "- 使用 `struct{}` 作为令牌类型，零字节开销。\n",
    "- `defer func() { <-semaphore }()` 保证释放令牌，避免出现无限等待。\n",
    "- 虽然一次性发起 1000 个 goroutine，但同时进入关键业务区的最多只有 100 个，实现资源保护。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// maxConcurrent 表示系统允许的最大并发请求数。\n",
    "// 这里设置为 100，对应“只能同时处理 100 个请求”的需求。\n",
    "const maxConcurrent = 100\n",
    "\n",
    "// semaphore 作为信号量，容量等于最大并发。\n",
    "// 发送元素表示占用一个并发名额，接收元素表示释放。\n",
    "var semaphore = make(chan struct{}, maxConcurrent)\n",
    "\n",
    "// mockHandler 模拟接口处理逻辑。\n",
    "// 每次调用前向 semaphore 写入一个空结构体，占用名额；处理结束后释放。\n",
    "func mockHandler(id int) {\n",
    "\t// 1. 获取令牌（若已满则阻塞，自动形成队列）\n",
    "\tsemaphore <- struct{}{}\n",
    "\t// 2. 确保函数退出时释放令牌，避免因 panic 或提前返回导致资源泄漏\n",
    "\tdefer func() { <-semaphore }()\n",
    "\n",
    "\tfmt.Printf(\"[req-%d] start\\n\", id)\n",
    "\t// 3. 模拟耗时操作：例如数据库查询、复杂计算等\n",
    "\ttime.Sleep(3 * time.Second)\n",
    "\tfmt.Printf(\"[req-%d] done\\n\", id)\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst totalRequests = 1000 // 模拟 1000 个并发用户请求\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(totalRequests)\n",
    "\n",
    "\tstart := time.Now()\n",
    "\tfor i := 0; i < totalRequests; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tmockHandler(id)\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\telapsed := time.Since(start)\n",
    "\tfmt.Printf(\"全部请求处理完毕，总耗时：%v\\n\", elapsed)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### HTTP 接口限流 + 等待超时处理"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 通过 `context.WithTimeout` 控制获取令牌的等待时间，兼顾资源保护与用户体验。\n",
    "- 业务逻辑内应尽量少用 `time.Sleep`，应替换为实际操作；此处仅用于演示。\n",
    "- 可结合 metrics 记录 `len(semaphore)`、等待时长、拒绝率等指标，进一步调优。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"context\"\n",
    "\t\"fmt\"\n",
    "\t\"log\"\n",
    "\t\"net/http\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "const (\n",
    "\tmaxConcurrent = 50          // 允许同时处理的请求数\n",
    "\twaitTimeout   = 2 * time.Second // 获取令牌的最大等待时间\n",
    ")\n",
    "\n",
    "var semaphore = make(chan struct{}, maxConcurrent)\n",
    "\n",
    "// acquire 尝试在给定上下文中获取令牌。\n",
    "// 成功返回 true，失败（超时或取消）返回 false。\n",
    "func acquire(ctx context.Context) bool {\n",
    "\tselect {\n",
    "\tcase semaphore <- struct{}{}:\n",
    "\t\treturn true\n",
    "\tcase <-ctx.Done():\n",
    "\t\treturn false\n",
    "\t}\n",
    "}\n",
    "\n",
    "// release 归还令牌，必须在业务逻辑结束后调用。\n",
    "func release() {\n",
    "\tselect {\n",
    "\tcase <-semaphore:\n",
    "\t\t// 正常释放\n",
    "\tdefault:\n",
    "\t\t// 理论上不会发生，防御性处理\n",
    "\t\tlog.Println(\"warn: release called without acquire\")\n",
    "\t}\n",
    "}\n",
    "\n",
    "// businessLogic 模拟接口的实际工作，耗时 500ms。\n",
    "func businessLogic(ctx context.Context, w http.ResponseWriter) error {\n",
    "\tselect {\n",
    "\tcase <-ctx.Done():\n",
    "\t\treturn ctx.Err()\n",
    "\tcase <-time.After(500 * time.Millisecond):\n",
    "\t\tfmt.Fprintln(w, \"OK: handled request\")\n",
    "\t\treturn nil\n",
    "\t}\n",
    "}\n",
    "\n",
    "// rateLimitedHandler 包装真实 handler，实现并发限制 + 获取令牌超时处理。\n",
    "func rateLimitedHandler(w http.ResponseWriter, r *http.Request) {\n",
    "\tctx, cancel := context.WithTimeout(r.Context(), waitTimeout)\n",
    "\tdefer cancel()\n",
    "\n",
    "\tif ok := acquire(ctx); !ok {\n",
    "\t\thttp.Error(w, \"Too Many Requests (waiting timeout)\", http.StatusTooManyRequests)\n",
    "\t\treturn\n",
    "\t}\n",
    "\tdefer release()\n",
    "\n",
    "\tif err := businessLogic(ctx, w); err != nil {\n",
    "\t\thttp.Error(w, err.Error(), http.StatusGatewayTimeout)\n",
    "\t\treturn\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tmux := http.NewServeMux()\n",
    "\tmux.HandleFunc(\"/api/do\", rateLimitedHandler)\n",
    "\n",
    "\tserver := &http.Server{\n",
    "\t\tAddr:         \":8080\",\n",
    "\t\tHandler:      mux,\n",
    "\t\tReadTimeout:  5 * time.Second,\n",
    "\t\tWriteTimeout: 10 * time.Second,\n",
    "\t}\n",
    "\n",
    "\tlog.Println(\"HTTP server started on http://localhost:8080/api/do\")\n",
    "\tif err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {\n",
    "\t\tlog.Fatalf(\"listen: %v\\n\", err)\n",
    "\t}\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 发送 200 个并发请求，观察有些请求被 429 拒绝\n",
    "\n",
    "```shell\n",
    "hey -n 200 -c 200 http://localhost:8080/api/do\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 103 用channel限制协程的总数"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 关注维度 | 核心要点                                                     | Go 实现技巧                                                  | 易忽略风险                                                |\n",
    "| -------- | ------------------------------------------------------------ | ------------------------------------------------------------ | --------------------------------------------------------- |\n",
    "| 限制动机 | 防止一次性启动几十万 Goroutine 导致内存膨胀、调度开销、FD/句柄耗尽 | 统一入口创建协程，设定硬性上限                               | 未统一入口会出现“漏网之鱼”依旧无限制增长                  |\n",
    "| 控制模型 | 使用带缓冲的 `chan struct{}` 充当“并发令牌”                  | `sem := make(chan struct{}, limit)`，创建前写入，结束后读出  | 忘记释放令牌会造成阻塞 + 死锁                             |\n",
    "| 监控手段 | 周期性打印 `runtime.NumGoroutine()` 观察实时数量             | `time.NewTicker` + 独立监控协程                              | 记得 `defer ticker.Stop()`，避免资源泄漏                  |\n",
    "| 封装范式 | `GoroutineLimiter.Run(func())` 统一封装 `go` 关键字          | 结构体持有 `limit` 与 `chan struct{}`，Run 内部负责写入/释放 | 不可在 `Run` 中 `go` 前 return；函数 panic 需确保令牌释放 |\n",
    "| 调优策略 | limit 依据 CPU / 内存 / 下游资源设定                         | 结合监控判断是否需要调高或调低                               | limit 过低会降低吞吐；过高起不到保护作用                  |\n",
    "| 扩展能力 | 可叠加上下文、超时、任务队列等功能                           | `RunContext`、`TryRun` 等变体                                | 需避免嵌套 limiter 造成死锁链                             |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 最小可运行 Goroutine 限制器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 任何调用 `limiter.Run` 的地方都受限于 `maxAlive`。\n",
    "- 若将 `maxAlive` 改为 `totalJobs`，行为就与无限制相同；观察 monitor 输出即可体会限制效果。\n",
    "- `runtime.NumGoroutine()` 会包含 main、监控 goroutine，本例峰值 ≈ `maxAlive + 2`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// GoroutineLimiter 使用带缓冲 channel 控制并发 Goroutine 数量。\n",
    "type GoroutineLimiter struct {\n",
    "\tlimit int\n",
    "\ttokens chan struct{}\n",
    "}\n",
    "\n",
    "// NewGoroutineLimiter 构造函数，limit 决定最大并发 Goroutine 总数。\n",
    "func NewGoroutineLimiter(limit int) *GoroutineLimiter {\n",
    "\treturn &GoroutineLimiter{\n",
    "\t\tlimit:  limit,\n",
    "\t\ttokens: make(chan struct{}, limit),\n",
    "\t}\n",
    "}\n",
    "\n",
    "// Run 将传入的函数 f 包装成受限的 goroutine。\n",
    "// 只要 tokens 被占满，新的 Run 调用会阻塞，直到有 goroutine 结束。\n",
    "func (g *GoroutineLimiter) Run(f func()) {\n",
    "\tg.tokens <- struct{}{} // 占用名额（当 channel 满时阻塞）\n",
    "\tgo func() {\n",
    "\t\tdefer func() {\n",
    "\t\t\t<-g.tokens // 释放名额，必须放在 defer 中确保 panic 也能释放\n",
    "\t\t}()\n",
    "\t\tf()\n",
    "\t}()\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tconst (\n",
    "\t\ttotalJobs = 10000 // 总任务量\n",
    "\t\tmaxAlive  = 100   // 最大允许同时运行的 goroutine 数\n",
    "\t)\n",
    "\n",
    "\tlimiter := NewGoroutineLimiter(maxAlive)\n",
    "\n",
    "\t// 启动监控协程，每秒输出当前 Goroutine 数量（包含 main/监控/工作协程）\n",
    "\tticker := time.NewTicker(time.Second)\n",
    "\tdefer ticker.Stop()\n",
    "\tgo func() {\n",
    "\t\tfor range ticker.C {\n",
    "\t\t\tfmt.Printf(\"[monitor] 当前 Goroutine 数：%d\\n\", runtime.NumGoroutine())\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(totalJobs)\n",
    "\n",
    "\tjob := func(id int) {\n",
    "\t\tdefer wg.Done()\n",
    "\t\t// 模拟耗时任务：例如 I/O、计算等\n",
    "\t\ttime.Sleep(3 * time.Second)\n",
    "\t}\n",
    "\n",
    "\tstart := time.Now()\n",
    "\tfor i := 0; i < totalJobs; i++ {\n",
    "\t\ttaskID := i\n",
    "\t\tlimiter.Run(func() { job(taskID) })\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\tfmt.Printf(\"全部任务耗时：%v\\n\", time.Since(start))\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 限制协程 + 任务队列 + 超时保护"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `RunContext` 让我们可以在高峰时快速拒绝任务而不是一直排队，防止无限等待。\n",
    "- `TryRun` 适合非阻塞尝试，例如指标上报或次要任务。\n",
    "- `workerPool` 示范：任务到达过快时会触发 `context.DeadlineExceeded`，“丢弃”或记录告警。\n",
    "- 监控 goroutine 让我们能直观看到受限后 Goroutine 数量保持平稳。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"context\"\n",
    "\t\"errors\"\n",
    "\t\"fmt\"\n",
    "\t\"runtime\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// GoroutineLimiter 支持 TryRun / RunContext 等扩展接口。\n",
    "type GoroutineLimiter struct {\n",
    "\tlimit  int\n",
    "\ttokens chan struct{}\n",
    "}\n",
    "\n",
    "func NewGoroutineLimiter(limit int) *GoroutineLimiter {\n",
    "\treturn &GoroutineLimiter{\n",
    "\t\tlimit:  limit,\n",
    "\t\ttokens: make(chan struct{}, limit),\n",
    "\t}\n",
    "}\n",
    "\n",
    "// RunContext 在 ctx 允许的前提下运行任务，如果在 ctx 取消前未获得 token，则返回错误。\n",
    "func (g *GoroutineLimiter) RunContext(ctx context.Context, f func()) error {\n",
    "\tselect {\n",
    "\tcase g.tokens <- struct{}{}:\n",
    "\t\tgo func() {\n",
    "\t\t\tdefer func() { <-g.tokens }()\n",
    "\t\t\tf()\n",
    "\t\t}()\n",
    "\t\treturn nil\n",
    "\tcase <-ctx.Done():\n",
    "\t\treturn ctx.Err()\n",
    "\t}\n",
    "}\n",
    "\n",
    "// TryRun 尝试立即运行任务；若令牌被占满则返回 false。\n",
    "func (g *GoroutineLimiter) TryRun(f func()) bool {\n",
    "\tselect {\n",
    "\tcase g.tokens <- struct{}{}:\n",
    "\t\tgo func() {\n",
    "\t\t\tdefer func() { <-g.tokens }()\n",
    "\t\t\tf()\n",
    "\t\t}()\n",
    "\t\treturn true\n",
    "\tdefault:\n",
    "\t\treturn false\n",
    "\t}\n",
    "}\n",
    "\n",
    "// workerPool 演示如何结合 limiter 构建任务队列（生产者-消费者）。\n",
    "type workerPool struct {\n",
    "\tlimiter *GoroutineLimiter\n",
    "\tqueue   chan int\n",
    "\twg      sync.WaitGroup\n",
    "}\n",
    "\n",
    "func newWorkerPool(limit, queueSize int) *workerPool {\n",
    "\treturn &workerPool{\n",
    "\t\tlimiter: NewGoroutineLimiter(limit),\n",
    "\t\tqueue:   make(chan int, queueSize),\n",
    "\t}\n",
    "}\n",
    "\n",
    "// start 启动后台协程消费队列任务，并接受 context 控制退出。\n",
    "func (p *workerPool) start(ctx context.Context) {\n",
    "\tp.wg.Add(1)\n",
    "\tgo func() {\n",
    "\t\tdefer p.wg.Done()\n",
    "\t\tfor {\n",
    "\t\t\tselect {\n",
    "\t\t\tcase job, ok := <-p.queue:\n",
    "\t\t\t\tif !ok {\n",
    "\t\t\t\t\treturn\n",
    "\t\t\t\t}\n",
    "\t\t\t\tjobID := job\n",
    "\t\t\t\t// 使用 RunContext，若 200ms 内无法获取令牌视为过载，丢弃任务\n",
    "\t\t\t\ttaskCtx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)\n",
    "\t\t\t\terr := p.limiter.RunContext(taskCtx, func() {\n",
    "\t\t\t\t\tdefer cancel()\n",
    "\t\t\t\t\tdoWork(jobID)\n",
    "\t\t\t\t})\n",
    "\t\t\t\tif errors.Is(err, context.DeadlineExceeded) {\n",
    "\t\t\t\t\tfmt.Printf(\"[drop] job %d discarded: limiter busy\\n\", jobID)\n",
    "\t\t\t\t\tcancel()\n",
    "\t\t\t\t}\n",
    "\t\t\tcase <-ctx.Done():\n",
    "\t\t\t\treturn\n",
    "\t\t\t}\n",
    "\t\t}\n",
    "\t}()\n",
    "}\n",
    "\n",
    "func (p *workerPool) submit(job int) {\n",
    "\tp.queue <- job\n",
    "}\n",
    "\n",
    "func (p *workerPool) stop() {\n",
    "\tclose(p.queue)\n",
    "\tp.wg.Wait()\n",
    "}\n",
    "\n",
    "func doWork(id int) {\n",
    "\tfmt.Printf(\"[worker] start job %d\\n\", id)\n",
    "\ttime.Sleep(150 * time.Millisecond) // 模拟处理延迟\n",
    "\tfmt.Printf(\"[worker] finish job %d\\n\", id)\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\tctx, cancel := context.WithCancel(context.Background())\n",
    "\tdefer cancel()\n",
    "\n",
    "\tpool := newWorkerPool(5, 20) // 最多 5 个并发 worker，队列容量 20\n",
    "\tpool.start(ctx)\n",
    "\n",
    "\t// 监控协程：每秒打印 Goroutine 数，观察 limiter 效果\n",
    "\tticker := time.NewTicker(time.Second)\n",
    "\tdefer ticker.Stop()\n",
    "\tgo func() {\n",
    "\t\tfor range ticker.C {\n",
    "\t\t\tfmt.Printf(\"[monitor] goroutines=%d\\n\", runtime.NumGoroutine())\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 向队列发送 40 个任务\n",
    "\tfor i := 0; i < 40; i++ {\n",
    "\t\tpool.submit(i)\n",
    "\t}\n",
    "\n",
    "\ttime.Sleep(5 * time.Second) // 等待任务处理\n",
    "\tpool.stop()\n",
    "\tfmt.Println(\"全部任务处理流程结束\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 104 select多路监听"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 关注维度 | 核心要点                                                     | Go 实现技巧                                         | 易错 / 易忽视事项                                            |\n",
    "| -------- | ------------------------------------------------------------ | --------------------------------------------------- | ------------------------------------------------------------ |\n",
    "| 监听能力 | `select` 可以同时等待多个 `channel` 的读写事件               | `select { case <-ch1:… case ch2<-val:… default:… }` | 与 `switch` 不同：`select` 只能操作 channel，且每轮仅执行一个 case |\n",
    "| 阻塞行为 | 没有 `default` 的 `select` 会阻塞直到任一分支就绪；有 `default` 时若都不就绪立即执行 `default` | 使用 `default` 实现**非阻塞**尝试或兜底逻辑         | 误用 `default` 会导致忙轮询；需配合 `time.Sleep` 或其他节流手段 |\n",
    "| 随机选择 | 多个 case 同时满足时，Go runtime 会随机挑选，提高公平性      | 利用随机特性实现负载均衡                            | 不保证严格轮询，若需优先级需显式设计                         |\n",
    "| 退出控制 | 可借助 `label` + `break` / `continue` 精准控制外层循环；也可通过关闭 channel 或上下文取消 | `break label` 跳出多层循环；`return` 直接退出函数   | label 需紧贴 `for` / `switch` / `select` 声明，否则编译错误  |\n",
    "| 兜底清理 | 在退出前使用 `select` + `default` 非阻塞地清理残留数据       | `select { case v := <-ch: ... default: }`           | 避免退出时阻塞在空 channel 上                                |\n",
    "| 协程保活 | `select {}` 可永久阻塞主协程，确保后台 goroutine 继续运行    | 或使用 `<-make(chan struct{})`                      | 若所有 goroutine 结束，`select {}` 会导致 `fatal error: all goroutines are asleep - deadlock!` |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 多路监听 + 条件退出 + 兜底清理"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `default` 分支避免阻塞\n",
    "- 使用标签 `AB` 精准跳出外层 `for` 循环，与文档中 `break AB` 一致。\n",
    "- 最终兜底 `select` 保证退出时不会因为空 channel 而阻塞。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[ch1] value=8442927101217108631\n",
      "[ch2] value=110\n",
      "[ch1] value=5708867375591527038\n",
      "[ch2] value=75\n",
      "[ch2] value=30\n",
      "byte 值小于 40，准备退出监听循环\n",
      "[cleanup] ch1 残留数据：9116386743307336773\n",
      "主流程结束\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"fmt\"\n",
    "\t\"math/rand\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\t// 设定两个带缓冲的 channel，分别传输 int 与 byte。\n",
    "\tch1 := make(chan int, 1000)\n",
    "\tch2 := make(chan byte, 1000)\n",
    "\n",
    "\t// 启动写入 goroutine：随机等待 0~1000ms 后向 ch1 写入随机整数。\n",
    "\tgo func() {\n",
    "\t\tfor {\n",
    "\t\t\t// 随机等待，模拟不确定的数据到达时间。\n",
    "\t\t\ttime.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)\n",
    "\t\t\tch1 <- rand.Int()\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 向 ch2 写入随机 byte 值的 goroutine。\n",
    "\tgo func() {\n",
    "\t\tfor {\n",
    "\t\t\ttime.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)\n",
    "\t\t\tch2 <- byte(rand.Int())\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\t// 使用标签以便在 select 满足条件时跳出最外层循环。\n",
    "AB:\n",
    "\tfor {\n",
    "\t\t\t\t\t// 节流：避免忙轮询，也方便观察输出节奏。\n",
    "\t\ttime.Sleep(time.Second)\n",
    "\n",
    "\t\tselect {\n",
    "\t\tcase v1 := <-ch1:\n",
    "\t\t\t// 当 ch1 有数据时优先读取并打印。\n",
    "\t\t\tfmt.Printf(\"[ch1] value=%d\\n\", v1)\n",
    "\n",
    "\t\tcase v2 := <-ch2:\n",
    "\t\t\tfmt.Printf(\"[ch2] value=%d\\n\", v2)\n",
    "\t\t\tif v2 < 40 {\n",
    "\t\t\t\t// 当满足业务条件时跳出外层循环，演示 label + break 用法。\n",
    "\t\t\t\tfmt.Println(\"byte 值小于 40，准备退出监听循环\")\n",
    "\t\t\t\tbreak AB\n",
    "\t\t\t}\n",
    "\n",
    "\t\tdefault:\n",
    "\t\t\t// 当所有 channel 暂无数据时立即执行 default 分支，实现非阻塞“兜底”逻辑。\n",
    "\t\t\tfmt.Println(\"[default] 暂无数据，继续等待\")\n",
    "\t\t}\n",
    "\t}\n",
    "\n",
    "\t// 退出监听前，非阻塞地尝试读取 ch1 是否还有残留数据，避免遗留消息。\n",
    "\tselect {\n",
    "\tcase leftover := <-ch1:\n",
    "\t\tfmt.Printf(\"[cleanup] ch1 残留数据：%d\\n\", leftover)\n",
    "\tdefault:\n",
    "\t\tfmt.Println(\"[cleanup] ch1 无残留数据，安全退出\")\n",
    "\t}\n",
    "\n",
    "\tfmt.Println(\"主流程结束\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### select + 超时 / 停止信号 / 保活阻塞"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 演示 `select` 搭配 `context` 实现超时，避免无限等待某个 channel。\n",
    "- 使用关闭的 `stopCh` 作为广播信号（`close(stopCh)` 会让 `<-stopCh` 立即返回零值）。\n",
    "- 末尾的 `select {}` 复用了文档中 `SelectBlock` 的思路，让后台协程得以持续运行。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"context\"\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "func main() {\n",
    "\tdataCh := make(chan string)\n",
    "\tstopCh := make(chan struct{})\n",
    "\n",
    "\t// 模拟后台生产者：2 秒后发送一条数据。\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(2 * time.Second)\n",
    "\t\tdataCh <- \"important payload\"\n",
    "\t}()\n",
    "\n",
    "\t// 模拟“取消信号”：4 秒后关闭 stopCh，代表系统被关闭。\n",
    "\tgo func() {\n",
    "\t\ttime.Sleep(4 * time.Second)\n",
    "\t\tclose(stopCh)\n",
    "\t}()\n",
    "\n",
    "\tctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)\n",
    "\tdefer cancel()\n",
    "\n",
    "\tselect {\n",
    "\tcase msg := <-dataCh:\n",
    "\t\tfmt.Println(\"收到数据：\", msg)\n",
    "\n",
    "\tcase <-stopCh:\n",
    "\t\t// stopCh 被关闭或发送信号时触发，优先级与其他 case 随机。\n",
    "\t\tfmt.Println(\"收到停止信号，立即退出\")\n",
    "\t\treturn\n",
    "\n",
    "\tcase <-ctx.Done():\n",
    "\t\t// 利用 context 实现超时控制，避免无限等待。\n",
    "\t\tfmt.Println(\"等待超时：\", ctx.Err())\n",
    "\t\treturn\n",
    "\t}\n",
    "\n",
    "\t// 保活示例：阻塞主协程使后台服务持续运行（与文档中 SelectBlock 类似）。\n",
    "\tfmt.Println(\"演示 select{} 阻塞主协程，防止守护协程提前退出\")\n",
    "\tgo func() {\n",
    "\t\tfor {\n",
    "\t\t\ttime.Sleep(time.Second)\n",
    "\t\t\tfmt.Println(\"后台协程仍在运行...\")\n",
    "\t\t}\n",
    "\t}()\n",
    "\n",
    "\tselect {} // 永久阻塞 main，使上面的后台协程持续输出。\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 105 不使用 `sync.Once` 的单例模式"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 关注维度     | 核心要点                                                    | 推荐做法                                             | 易忽略风险                                        |\n",
    "| ------------ | ----------------------------------------------------------- | ---------------------------------------------------- | ------------------------------------------------- |\n",
    "| 适用场景     | 配置文件、数据库连接、第三方 SDK 客户端等**高成本创建**对象 | 将创建逻辑集中在单例工厂函数                         | 若直接在业务中频繁读取/初始化，会重复消耗 I/O/CPU |\n",
    "| 基础思路     | 通过**包级私有全局变量**保存实例                            | `var cfg *config`（不可导出）                        | 若变量可导出，外部可绕过工厂函数                  |\n",
    "| 并发安全     | 多 goroutine 同时初始化需加锁保护                           | 使用 `sync.Mutex` + 双重检查锁定（DCL）              | 未加锁时会出现“判空后同时初始化”问题              |\n",
    "| 双重检查锁定 | “判空 → 加锁 → 再判空 → 初始化”                             | 锁仅在实例尚未创建时参与竞争，减小性能开销           | 忘记第二次判空会导致重复初始化                    |\n",
    "| 访问入口     | 统一的 `GetInstance()` 函数                                 | 函数内包含判空、加锁与返回                           | 在包外禁止直接 `&config{}` 创建                   |\n",
    "| 测试策略     | 并发场景下验证只初始化一次；校验字段读取正确                | `sync.WaitGroup` + 多 goroutine 调用 `GetInstance()` | 缺乏测试难以发现数据竞态或初始化重复问题          |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "要点说明\n",
    "\n",
    "1. **应用场景**：配置文件读取、数据库连接等昂贵操作只需执行一次，避免重复消耗资源。\n",
    "2. **基础实现流程**：\n",
    "   - 私有结构体（小写首字母）防止外部直接实例化。\n",
    "   - 包级变量缓存实例，`GetConfig()`/`NewServiceConfig()` 作为唯一访问入口。\n",
    "3. **并发安全**：\n",
    "   - 多 goroutine 同时访问时若缺少保护，会重复初始化。\n",
    "   - 双重检查锁定（DCL）在保证安全的同时减少锁竞争。\n",
    "4. **测试建议**：\n",
    "   - 通过多 goroutine 调用验证初始化仅执行一次，可观察日志/对象指针地址。\n",
    "5. **性能优化**：\n",
    "   - 锁操作置于判空内部；利用 `atomic.Pointer` 再次降低锁竞争。\n",
    "6. **与 `sync.Once` 对比**：\n",
    "   - `sync.Once` 封装了类似逻辑；当业务需自定义初始化流程或更细粒度控制时，可手动实现 DCL。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 双重检查锁定实现配置单例（无 sync.Once）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 多个 goroutine 输出的 config 指针地址相同，证明仅初始化一次"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[worker-4] config address=0xc00019a200, app=demo-app@v1.0.0\n",
      "[worker-1] config address=0xc00019a200, app=demo-app@v1.0.0\n",
      "[worker-0] config address=0xc00019a200, app=demo-app@v1.0.0\n",
      "[worker-2] config address=0xc00019a200, app=demo-app@v1.0.0\n",
      "[worker-3] config address=0xc00019a200, app=demo-app@v1.0.0\n"
     ]
    }
   ],
   "source": [
    "! cd 105/01 && go run ."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 原子指针 + 互斥锁实现懒加载单例（仍未使用 once）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 依旧未使用 sync.Once，通过 `atomic.Pointer + sync.Mutex` 避免重复初始化，适合对读取性能有更高要求的场景。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[g3] instance=0xc0000120d0, ping=pong from https://api.example.com\n",
      "[g2] instance=0xc0000120d0, ping=pong from https://api.example.com\n",
      "[g0] instance=0xc0000120d0, ping=pong from https://api.example.com\n",
      "[g1] instance=0xc0000120d0, ping=pong from https://api.example.com\n"
     ]
    }
   ],
   "source": [
    "! cd 105/02 && go run ."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 106 context 用法详解"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 关注维度                      | 核心要点                                                     | 推荐实践                                                     | 易忽视风险                                                   |\n",
    "| ----------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |\n",
    "| 接口组成 / `Context` 接口结构 | `Deadline() (time.Time, bool)`、`Done() <-chan struct{}`、`Err() error`、`Value(key any) any` 定义协程生命周期与元数据传递 | 所有公共逻辑优先将 `Context` 作为首个参数；通过 `context.Background()` / `context.TODO()` 获取根上下文 | 忽视 `ctx.Err()` 会错失取消/超时原因；`Done()` 信号触发后未及时退出造成泄漏 |\n",
    "| 根上下文                      | 根 Context 来源于标准库内部的 `emptyCtx`，对外仅暴露 `Background()` 与 `TODO()` | 生产代码使用 `Background()` 作为根；`TODO()` 用于暂未确定的占位场景 | 禁止自行构造 `emptyCtx`；长期保留 `TODO()` 会让上下文意图不明确 |\n",
    "| 生命周期控制                  | `Done()` 是只读通道，关闭代表取消或超时；`Err()` 报告关闭原因 | 在 `select` 中同时监听业务完成与 `ctx.Done()`，并在退出时记录 `Err()` | 未处理 `Done()` 信号容易导致 goroutine 泄漏或无故阻塞        |\n",
    "| 取消控制                      | `context.WithCancel` 返回 `(Context, cancel)`；显式调用 `cancel()` 关闭信号 | 始终 `defer cancel()` 保证资源释放；子协程使用 `select` 监听 `Done()` | 忘记调用 `cancel()` 会泄漏定时器/子协程；误以为 `cancel()` 会立即终止子协程 |\n",
    "| 超时控制                      | `WithTimeout` / `WithDeadline` 自动设置到期时间，超时即关闭 `Done()` | 利用 `select { case <-ctx.Done(): ... }` 第一时间响应超时；结合 `Err()` 判断状态 | 子 Context 实际超时时间不会超过父 Context 剩余时间；超时后 `Err()` 为 `context deadline exceeded` |\n",
    "| 值传递                        | `WithValue` 用于跨 API/协程传递请求级元数据（如请求 ID、用户角色） | 自定义 `key` 类型避免冲突；仅存储小型、不可变、线程安全的数据；读取时做好类型断言 | 滥用 `WithValue` 代替函数参数使耦合度增高；传入大对象或可变数据产生隐患 |\n",
    "| 调用链继承                    | 子 Context 共享父 Context 的 `Done` 信号与 `Value` 链路，寿命不超过父级 | 在微服务调用链中逐级透传剩余时间，防止下游无限等待耗尽资源   | 忽视父 Context 的超时限制会导致子请求提前取消；误设子超时大于父仍无效 |\n",
    "| 常见模式                      | `WithCancel` 控制协程退出；`WithTimeout` 做防御性超时；`WithValue` 透传元数据；所有协程监听 `Done()` | 搭配 `sync.WaitGroup` / `errgroup` 管理协程生命周期并统一错误处理 | 未取消 Context 或未处理 `Done()` 信号会导致 goroutine 泄漏、任务悬挂 |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "要点\n",
    "\n",
    "1. **Context 接口细节**\n",
    "   - `Deadline()`：返回“到期时间 + 是否设置”，用于链路传递剩余时间。\n",
    "   - `Done()`：返回只读 `<-chan struct{}` 信号通道。空结构体零开销，仅用于通知，不承载数据。\n",
    "   - `Err()`：区分 `context.Canceled`（手动取消）与 `context.DeadlineExceeded`（超时到期）。\n",
    "   - `Value()`：键值查找，建议定义独有 key 类型（例如 `type StringKey string`）防冲突。\n",
    "2. **`emptyCtx` 与根 Context**\n",
    "   - `emptyCtx` 是标准库内部空实现，通过全局变量 `background`, `todo` 暴露。\n",
    "   - 应使用 `context.Background()` 作为根 Context，`context.TODO()` 仅作为待定占位。\n",
    "3. **`WithCancel` 演示重点**\n",
    "   - 调用 `context.WithCancel(parent)` 获得 `(childCtx, cancel)`。\n",
    "   - `cancel()` 调用即关闭 `Done()` 管道，所有监听者立即获知取消信号。\n",
    "   - 常用于手动停止后台 worker、超时协程或响应外部信号。\n",
    "4. **`WithTimeout`/`WithDeadline` 演示重点**\n",
    "   - 与 `WithCancel` 相比多了自动到期功能；到期后自动触发 `Done()`。\n",
    "   - 一旦超时，`Err()` 返回 `context deadline exceeded`。\n",
    "   - 子 Context 的实际到期时间是父剩余时间与子设置时间的最小值。\n",
    "5. **Context 继承规则**\n",
    "   - 子 Context 生命周期不能超过父 Context（即 `Done` 链式传播）。\n",
    "   - 微服务调用链应把剩余时间传递给下游，防止调用无限等待。\n",
    "   - 示例：父剩 500ms，子设 1000ms → 实际以 500ms 为准；子设 100ms → 100ms。\n",
    "6. **`WithValue` 传递数据**\n",
    "   - 用于跨 API/协程传递请求级元信息（如请求 ID、用户身份）。\n",
    "   - 取值时需类型断言：`ctx.Value(key).(Type)`。\n",
    "   - 针对 HTTP、RPC 等场域，可在服务端与客户端链式共享上下文。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### WithCancel 手动取消后台任务"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `mockWork` 通过 `select` 监听 `ctx.Done()`，第一时间响应取消。\n",
    "- `cancel()` 必须在不再使用 Context 时调用，否则可能导致资源泄漏（如定时器未释放）。\n",
    "- `ctx.Err()` 返回具体原因，便于日志记录或指标采集。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[worker-1] processed batch\n",
      "[worker-1] processed batch\n",
      "[worker-0] processed batch\n",
      "[main] detect anomaly, cancel all workers\n",
      "[worker-1] exit: context canceled\n",
      "[worker-0] exit: context canceled\n",
      "[main] shutdown complete\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"context\"\n",
    "\t\"fmt\"\n",
    "\t\"math/rand\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// mockWork 模拟耗时任务：持续从数据源接收数据，直到收到取消信号。\n",
    "func mockWork(ctx context.Context, id int) {\n",
    "\tfor {\n",
    "\t\tselect {\n",
    "\t\tcase <-ctx.Done():\n",
    "\t\t\t// ctx.Done() 关闭时立即退出，并报告原因（取消 / 超时）\n",
    "\t\t\tfmt.Printf(\"[worker-%d] exit: %v\\n\", id, ctx.Err())\n",
    "\t\t\treturn\n",
    "\t\tcase <-time.After(time.Duration(200+rand.Intn(400)) * time.Millisecond):\n",
    "\t\t\tfmt.Printf(\"[worker-%d] processed batch\\n\", id)\n",
    "\t\t}\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\t// 背景根 Context：可作为整个进程的根节点。\n",
    "\troot := context.Background()\n",
    "\n",
    "\t// WithCancel 返回派生 Context 和取消函数。\n",
    "\tctx, cancel := context.WithCancel(root)\n",
    "\n",
    "\tfor i := 0; i < 2; i++ {\n",
    "\t\tgo mockWork(ctx, i)\n",
    "\t}\n",
    "\n",
    "\t// 主流程逻辑：例如监听外部信号、检测异常等，这里简化为定时器。\n",
    "\ttime.Sleep(750 * time.Millisecond)\n",
    "\tfmt.Println(\"[main] detect anomaly, cancel all workers\")\n",
    "\n",
    "\t// cancel() 关闭 ctx.Done() 信号，提醒所有子协程立刻退出。\n",
    "\tcancel()\n",
    "\n",
    "\t// 预留时间让后台协程完成清理。\n",
    "\ttime.Sleep(300 * time.Millisecond)\n",
    "\tfmt.Println(\"[main] shutdown complete\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### WithTimeout 自动超时 + Context 继承链"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 父 Context 设置 120ms，子 Context 即使要求 300ms，实际只剩父 Context 的剩余时间。\n",
    "- 各阶段 `Err()` 返回 `context deadline exceeded`，说明触发超时。\n",
    "- `defer cancel()` 是惯用写法，防止定时器资源泄漏。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[main] database error: context deadline exceeded\n",
      "[main] downstream error: context deadline exceeded\n",
      "[main] total elapsed: 120.535111ms\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"context\"\n",
    "\t\"fmt\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// queryDatabase 模拟数据库查询，耗时 150ms。\n",
    "func queryDatabase(ctx context.Context) error {\n",
    "\tselect {\n",
    "\tcase <-ctx.Done():\n",
    "\t\treturn ctx.Err() // 返回 context 的取消原因\n",
    "\tcase <-time.After(150 * time.Millisecond):\n",
    "\t\tfmt.Println(\"[DB] query finished\")\n",
    "\t\treturn nil\n",
    "\t}\n",
    "}\n",
    "\n",
    "// callDownstream 模拟调用下游服务，预期耗时 100ms。\n",
    "func callDownstream(ctx context.Context) error {\n",
    "\tselect {\n",
    "\tcase <-ctx.Done():\n",
    "\t\treturn ctx.Err()\n",
    "\tcase <-time.After(100 * time.Millisecond):\n",
    "\t\tfmt.Println(\"[RPC] downstream finished\")\n",
    "\t\treturn nil\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\troot := context.Background()\n",
    "\n",
    "\t// 根请求超时时间设置为 120ms。\n",
    "\tctx, cancel := context.WithTimeout(root, 120*time.Millisecond)\n",
    "\tdefer cancel() // 确保超时或返回前释放资源\n",
    "\n",
    "\tstart := time.Now()\n",
    "\n",
    "\t// 子请求：数据库查询试图设置 300ms 超时，但实际会被父 Context 的剩余时间限制。\n",
    "\tdbCtx, dbCancel := context.WithTimeout(ctx, 300*time.Millisecond)\n",
    "\tdefer dbCancel()\n",
    "\n",
    "\tif err := queryDatabase(dbCtx); err != nil {\n",
    "\t\tfmt.Println(\"[main] database error:\", err)\n",
    "\t}\n",
    "\n",
    "\t// 再次派生子 Context，要求 80ms。\n",
    "\trpcCtx, rpcCancel := context.WithTimeout(ctx, 80*time.Millisecond)\n",
    "\tdefer rpcCancel()\n",
    "\n",
    "\tif err := callDownstream(rpcCtx); err != nil {\n",
    "\t\tfmt.Println(\"[main] downstream error:\", err)\n",
    "\t}\n",
    "\n",
    "\tfmt.Printf(\"[main] total elapsed: %v\\n\", time.Since(start))\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### WithValue 透传请求元数据 + 并发协程共享 Context"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `WithValue` 仅推荐在“跨 API 边界传递少量元数据”的场景使用；不要当作全局变量或参数的替代品。\n",
    "- 读取时需要类型断言；未设置值时应提供安全的默认值。\n",
    "- Value 数据会随 Context 传递给所有下游协程。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025/09/30 21:34:54 [rid=req-12345][role=admin] worker-2 started\n",
      "2025/09/30 21:34:54 [rid=req-12345][role=admin] worker-0 started\n",
      "2025/09/30 21:34:54 [rid=req-12345][role=admin] worker-1 started\n",
      "2025/09/30 21:34:54 [rid=req-12345][role=admin] worker-0 finished work\n",
      "2025/09/30 21:34:55 [rid=req-12345][role=admin] worker-1 finished work\n",
      "2025/09/30 21:34:55 [rid=req-12345][role=admin] worker-2 cancelled: context deadline exceeded\n",
      "2025/09/30 21:34:55 [main] all workers finished\n"
     ]
    }
   ],
   "source": [
    "package main\n",
    "\n",
    "import (\n",
    "\t\"context\"\n",
    "\t\"fmt\"\n",
    "\t\"log\"\n",
    "\t\"sync\"\n",
    "\t\"time\"\n",
    ")\n",
    "\n",
    "// 定义专用 key 类型，避免与其他包的 key 冲突。\n",
    "type ctxKey string\n",
    "\n",
    "const (\n",
    "\tkeyRequestID ctxKey = \"requestID\"\n",
    "\tkeyUserRole  ctxKey = \"userRole\"\n",
    ")\n",
    "\n",
    "// logWithContext 演示如何在下游协程中读取上下文携带的值。\n",
    "func logWithContext(ctx context.Context, msg string) {\n",
    "\treqID, _ := ctx.Value(keyRequestID).(string) // 提供默认值以防断言失败\n",
    "\trole, _ := ctx.Value(keyUserRole).(string)\n",
    "\tlog.Printf(\"[rid=%s][role=%s] %s\\n\", reqID, role, msg)\n",
    "}\n",
    "\n",
    "// worker 模拟后台任务，在执行过程中也会关心取消信号。\n",
    "func worker(ctx context.Context, id int) {\n",
    "\tlogWithContext(ctx, fmt.Sprintf(\"worker-%d started\", id))\n",
    "\tselect {\n",
    "\tcase <-ctx.Done():\n",
    "\t\tlogWithContext(ctx, fmt.Sprintf(\"worker-%d cancelled: %v\", id, ctx.Err()))\n",
    "\tcase <-time.After(time.Duration(100+id*50) * time.Millisecond):\n",
    "\t\tlogWithContext(ctx, fmt.Sprintf(\"worker-%d finished work\", id))\n",
    "\t}\n",
    "}\n",
    "\n",
    "func main() {\n",
    "\troot := context.Background()\n",
    "\n",
    "\t// 使用 WithValue 链式添加元信息：请求 ID 与用户角色。\n",
    "\tctx := context.WithValue(root, keyRequestID, \"req-12345\")\n",
    "\tctx = context.WithValue(ctx, keyUserRole, \"admin\")\n",
    "\n",
    "\t// 设置整体请求超时 180ms。\n",
    "\tctx, cancel := context.WithTimeout(ctx, 180*time.Millisecond)\n",
    "\tdefer cancel()\n",
    "\n",
    "\tvar wg sync.WaitGroup\n",
    "\twg.Add(3)\n",
    "\n",
    "\tfor i := 0; i < 3; i++ {\n",
    "\t\tgo func(id int) {\n",
    "\t\t\tdefer wg.Done()\n",
    "\t\t\tworker(ctx, id)\n",
    "\t\t}(i)\n",
    "\t}\n",
    "\n",
    "\twg.Wait()\n",
    "\tlog.Println(\"[main] all workers finished\")\n",
    "}"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Go (gonb)",
   "language": "go",
   "name": "gonb"
  },
  "language_info": {
   "codemirror_mode": "",
   "file_extension": ".go",
   "mimetype": "text/x-go",
   "name": "go",
   "nbconvert_exporter": "",
   "pygments_lexer": "",
   "version": "go1.24.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
